shanxia2.0/backend/scripts/patch_pr3_api.py

#!/usr/bin/env python3
"""PR3 Task 2.4–2.7: API 层注入 user_info。"""
from __future__ import annotations

import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def _inject_user_info_after_tenant(body: str) -> str:
    if "user_info = crm_user_info(_user)" in body:
        return body
    lines = body.split("\n")
    for i, line in enumerate(lines):
        if "tenant_no = get_tenant_no(db, _user)" in line:
            indent = line[: len(line) - len(line.lstrip())]
            lines.insert(i + 1, f"{indent}user_info = crm_user_info(_user)")
            return "\n".join(lines)
    return body


def patch_route_file(path: Path, route_prefixes: tuple[str, ...]):
    c = path.read_text()
    parts = re.split(r'(@router\.(?:post|get|put|delete)\("[^"]+"\))', c)
    if len(parts) < 2:
        # register_* pattern
        parts = re.split(r'(    @router\.(?:post|get|put|delete)\("[^"]+"\))', c)
    out = [parts[0]]
    for i in range(1, len(parts), 2):
        dec = parts[i]
        body = parts[i + 1] if i + 1 < len(parts) else ""
        m = re.search(r'@router\.(?:post|get|put|delete)\("([^"]+)"\)', dec)
        rp = m.group(1) if m else ""
        if any(rp.startswith(p) for p in route_prefixes):
            body = _inject_user_info_after_tenant(body)
        out.append(dec)
        out.append(body)
    c = "".join(out)
    path.write_text(c)
    print(f"injected user_info blocks in {path.name}")


def patch_crm_py_calls():
    path = ROOT / "app/api/crm.py"
    c = path.read_text()

    # imports
    if "ensure_crm_contact_row_in_scope" not in c:
        c = c.replace(
            "    ensure_crm_owner_row_in_scope,\n)",
            "    ensure_crm_contact_row_in_scope,\n    ensure_crm_customer_linked_row_in_scope,\n    ensure_crm_owner_row_in_scope,\n)",
        )

    replacements = [
        # product categories
        (
            "rows = list_category_tree(\n            db,\n            tenant_no,\n            category_name=category_name,\n            status=status,\n        )",
            "rows = list_category_tree(\n            db,\n            tenant_no,\n            category_name=category_name,\n            status=status,\n            user_info=user_info,\n        )",
        ),
        (
            "obj = update_category(db, category_id, user_id=user_id, **data)",
            "obj = update_category(db, category_id, user_id=user_id, user_info=user_info, **data)",
        ),
        ("soft_delete_category(db, body.id)", "soft_delete_category(db, body.id, user_info=user_info)"),
        # products
        (
            "rows, total = list_products(\n            db,\n            tenant_no,\n            page=body.page,\n            page_size=body.pageSize,\n            keyword=body.keyword,\n            category_id=body.category_id,\n            status=body.status,\n        )",
            "rows, total = list_products(\n            db,\n            tenant_no,\n            page=body.page,\n            page_size=body.pageSize,\n            keyword=body.keyword,\n            category_id=body.category_id,\n            status=body.status,\n            user_info=user_info,\n        )",
        ),
        (
            "obj = get_product(db, body.id)\n        if not obj:\n            return error_response(404, \"产品不存在\")\n        return success_response(_product_row_to_dict(obj))",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        obj = get_product(db, body.id)\n        if not obj:\n            return error_response(404, \"产品不存在\")\n        ensure_crm_create_by_row_in_scope(db, obj, user_info, tenant_no, message=\"无权访问该产品\")\n        return success_response(_product_row_to_dict(obj))",
        ),
        (
            "obj = update_product(db, product_id, user_id=user_id, **data)",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        obj = update_product(db, product_id, user_id=user_id, user_info=user_info, **data)",
        ),
        (
            "soft_delete_product(db, pid)",
            "soft_delete_product(db, pid, user_info=user_info)",
        ),
        (
            "update_product(db, pid, user_id=user_id, status=body.status)",
            "update_product(db, pid, user_id=user_id, user_info=user_info, status=body.status)",
        ),
        (
            "options = list_product_options(\n            db,\n            tenant_no,\n            keyword=body.keyword,\n            category_id=body.category_id,\n        )",
            "options = list_product_options(\n            db,\n            tenant_no,\n            keyword=body.keyword,\n            category_id=body.category_id,\n            user_info=user_info,\n        )",
        ),
        # contact
        (
            "rows, total = list_contacts(\n            db, tenant_no, page=page, page_size=page_size,",
            "user_info = crm_user_info(_user)\n        rows, total = list_contacts(\n            db, tenant_no, page=page, page_size=page_size, user_info=user_info,",
        ),
        (
            "row = get_contact(db, contact_id)\n        if not row:\n            return error_response(404, \"联系人不存在\")\n        return success_response(contact_row_to_dict(row))",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        row = get_contact(db, contact_id)\n        if not row:\n            return error_response(404, \"联系人不存在\")\n        ensure_crm_contact_row_in_scope(db, row, user_info, tenant_no, message=\"无权访问该联系人\")\n        return success_response(contact_row_to_dict(row))",
        ),
        (
            "rows = list_contacts_by_customer(db, tenant_no, customer_id)",
            "user_info = crm_user_info(_user)\n        rows = list_contacts_by_customer(db, tenant_no, customer_id, user_info=user_info)",
        ),
        (
            "obj = create_contact(\n            db, tenant_no=tenant_no, user_id=user_id,",
            "user_info = crm_user_info(_user)\n        obj = create_contact(\n            db, tenant_no=tenant_no, user_id=user_id, user_info=user_info,",
        ),
        (
            "obj = update_contact(db, contact_id, user_id=user_id, **data)",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        obj = update_contact(db, contact_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info, **data)",
        ),
        (
            "soft_delete_contact(db, cid)",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        user_id = _user.get(\"user_id\", \"\") or _user.get(\"id\", \"\")\n            soft_delete_contact(db, cid, user_id=user_id, tenant_no=tenant_no, user_info=user_info)",
        ),
        (
            "obj = set_primary_contact(db, contact_id, user_id=user_id)",
            "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        obj = set_primary_contact(db, contact_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info)",
        ),
    ]

    for old, new in replacements:
        if old in c and new not in c:
            c = c.replace(old, new, 1)

    # Fix contact delete indentation if broken
    c = c.replace(
        "for cid in body.ids:\n        tenant_no = get_tenant_no(db, _user)",
        "tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        user_id = _user.get(\"user_id\", \"\") or _user.get(\"id\", \"\")\n        for cid in body.ids:",
    )

    # supplier / invoice - inject user_info in list handlers via route patch; add call params
    c = re.sub(
        r"list_supplier_info\(\s*\n\s*db, tenant_no,",
        "list_supplier_info(\n            db, tenant_no, user_info=user_info,",
        c,
    )
    c = re.sub(
        r"list_all_supplier_info\(\s*\n\s*db, tenant_no,",
        "list_all_supplier_info(\n            db, tenant_no, user_info=user_info,",
        c,
    )
    c = c.replace(
        "obj = update_supplier_info(\n            db,\n            supplier_id,\n            tenant_no=tenant_no,\n            user_id=user_id,",
        "obj = update_supplier_info(\n            db,\n            supplier_id,\n            tenant_no=tenant_no,\n            user_id=user_id,\n            user_info=user_info,",
    )
    c = c.replace(
        "soft_delete_supplier_info(\n            db,\n            sid,\n            tenant_no=tenant_no,\n            user_id=user_id,",
        "soft_delete_supplier_info(\n            db,\n            sid,\n            tenant_no=tenant_no,\n            user_id=user_id,\n            user_info=user_info,",
    )
    c = c.replace(
        "rows = list_invoice_info_by_customer(db, tenant_no, customer_id)",
        "rows = list_invoice_info_by_customer(db, tenant_no, customer_id, user_info=user_info)",
    )
    c = c.replace(
        "obj = create_invoice_info(\n            db,\n            tenant_no=tenant_no,\n            user_id=user_id,",
        "obj = create_invoice_info(\n            db,\n            tenant_no=tenant_no,\n            user_id=user_id,\n            user_info=user_info,",
    )
    c = c.replace(
        "obj = update_invoice_info(\n            db,\n            invoice_id,\n            tenant_no=tenant_no,\n            user_id=user_id,",
        "obj = update_invoice_info(\n            db,\n            invoice_id,\n            tenant_no=tenant_no,\n            user_id=user_id,\n            user_info=user_info,",
    )
    c = c.replace(
        "soft_delete_invoice_info(db, invoice_id, user_id=user_id)",
        "soft_delete_invoice_info(db, invoice_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info)",
    )
    c = c.replace(
        "obj = set_default_invoice_info(\n            db,\n            invoice_id,\n            customer_id,\n            tenant_no=tenant_no,\n            user_id=user_id,",
        "obj = set_default_invoice_info(\n            db,\n            invoice_id,\n            customer_id,\n            tenant_no=tenant_no,\n            user_id=user_id,\n            user_info=user_info,",
    )

    # product list/export need user_info at start of try
    c = c.replace(
        '"""分页查询产品列表"""\n    try:\n        tenant_no = get_tenant_no(db, _user)\n        rows, total = list_products(',
        '"""分页查询产品列表"""\n    try:\n        tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        rows, total = list_products(',
    )
    c = c.replace(
        '"""获取产品分类树"""\n    try:\n        tenant_no = get_tenant_no(db, _user)',
        '"""获取产品分类树"""\n    try:\n        tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)',
    )
    c = c.replace(
        '"""删除产品（支持单条 body.id 和批量 body.ids）"""\n    try:\n        ids_to_delete',
        '"""删除产品（支持单条 body.id 和批量 body.ids）"""\n    try:\n        tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        ids_to_delete',
    )
    c = c.replace(
        '"""批量上架/下架"""\n    try:\n        user_id = _user.get("user_id", "") or _user.get("id", "")',
        '"""批量上架/下架"""\n    try:\n        tenant_no = get_tenant_no(db, _user)\n        user_info = crm_user_info(_user)\n        user_id = _user.get("user_id", "") or _user.get("id", "")',
    )

    path.write_text(c)
    print("patched crm.py calls")


def patch_quote_routes():
    path = ROOT / "app/api/crm_quote_routes.py"
    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",
        )

    # inject user_info in each handler try block after tenant_no
    parts = re.split(r'(    @router\.post\("[^"]+"\))', c)
    out = [parts[0]]
    for i in range(1, len(parts), 2):
        dec, body = parts[i], parts[i + 1] if i + 1 < len(parts) else ""
        body = _inject_user_info_after_tenant(body)
        out.extend([dec, body])
    c = "".join(out)

    # pass user_info to crud calls
    c = re.sub(
        r"list_quotes\(\s*\n\s*db,\s*\n\s*tenant_no,",
        "list_quotes(\n                db,\n                tenant_no,\n                user_info=user_info,",
        c,
    )
    c = re.sub(
        r"get_quote_detail\(\s*\n\s*db,",
        "get_quote_detail(\n                db,",
        c,
    )
    c = c.replace(
        "tenant_no=tenant_no,\n            )",
        "tenant_no=tenant_no,\n                user_info=user_info,\n            )",
    )
    # Too broad - fix manually with targeted replaces
    path.write_text(c)
    print("patched crm_quote_routes.py (partial)")


def patch_contract_routes():
    path = ROOT / "app/api/crm_contract_routes.py"
    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",
        )
    parts = re.split(r'(    @router\.(?:get|post|put|delete)\("[^"]+"\))', c)
    out = [parts[0]]
    for i in range(1, len(parts), 2):
        dec, body = parts[i], parts[i + 1] if i + 1 < len(parts) else ""
        body = _inject_user_info_after_tenant(body)
        out.extend([dec, body])
    c = "".join(out)
    c = re.sub(
        r"list_contracts\(\s*\n\s*db,\s*\n\s*tenant_no,",
        "list_contracts(\n                db,\n                tenant_no,\n                user_info=user_info,",
        c,
    )
    c = re.sub(
        r"get_contract_detail\(\s*\n\s*db,",
        "get_contract_detail(\n                db,",
        c,
    )
    path.write_text(c)
    print("patched crm_contract_routes.py (partial)")


if __name__ == "__main__":
    patch_route_file(
        ROOT / "app/api/crm.py",
        (
            "/product-categories/",
            "/products/",
            "/contact/",
            "/supplierInfo/",
            "/invoice-info/",
        ),
    )
    patch_crm_py_calls()
    patch_quote_routes()
    patch_contract_routes()