#!/usr/bin/env python3 """为 quote/contract 路由 CRUD 调用补全 user_info=user_info。""" from __future__ import annotations import re from pathlib import Path ROOT = Path(__file__).resolve().parents[1] QUOTE_FUNCS = ( "get_quote_detail", "update_quote", "soft_delete_quotes", "submit_quote", "save_template", "get_template", "generate_pdf", "generate_share", "accept_quote", "reject_quote", "to_contract", "create_new_version", "query_versions", "withdraw_quote_approval", "invalid_quote", "statistics_overview", "statistics_status", "statistics_trend", "list_quote_approvals", "approve_quote", "withdraw_approval", "refresh_approver_names", "list_quote_products", "insert_quote_product", "update_quote_product", "soft_delete_quote_products", "list_share_logs", "share_log_statistics", ) CONTRACT_FUNCS = ( "get_contract_detail", "get_contract", "update_contract", "soft_delete_contracts", "submit_contract_approval", "withdraw_contract_approval", "sign_contract", "start_execute", "finish_contract", "invalid_contract", "renew_contract", "batch_assign_owner", "batch_submit_contract_approval", "upload_attachments", "upload_attachments_with_category", "upload_file", "save_template", "get_template", "download_contract_pdf", "create_from_quote", "create_from_opportunity", "export_contracts", "query_expiring", "statistics_overview", "statistics_by_status", "statistics_by_type", "statistics_trend", "list_contract_approvals", "approve_contract_by_contract_id", "list_contract_changes", "get_contract_change_detail", "update_contract_change", "soft_delete_contract_changes", "list_contract_templates", "update_contract_template", "soft_delete_contract_templates", ) def _add_user_info_kwarg(content: str, func_names: tuple[str, ...]) -> str: for fn in func_names: pattern = rf"({re.escape(fn)}\([^)]*?)(\n\s*\))" while True: m = re.search(pattern, content, re.DOTALL) if not m: break call = m.group(1) if "user_info=" in call: content = content[: m.start()] + call + m.group(2) + content[m.end() :] break new_call = call.rstrip() + ",\n user_info=user_info," content = content[: m.start()] + new_call + m.group(2) + content[m.end() :] return content def patch_file(rel: str, funcs: tuple[str, ...]): path = ROOT / rel c = path.read_text() if "crm_user_info" not in c: c = c.replace( "from app.core.tenant_context import get_tenant_no\n", "from app.core.tenant_context import get_tenant_no\n" "from app.crud.crm.data_scope_helpers import crm_user_info\n", ) c2 = _add_user_info_kwarg(c, funcs) path.write_text(c2) print(f"patched {rel}") if __name__ == "__main__": patch_file("app/api/crm_quote_routes.py", QUOTE_FUNCS) patch_file("app/api/crm_contract_routes.py", CONTRACT_FUNCS) sub = ROOT / "app/api/crm_contract_sub_routes.py" if sub.exists(): patch_file("app/api/crm_contract_sub_routes.py", CONTRACT_FUNCS)