AI 写出的代码，离生产还差什么？——从 Demo 到稳定的五道坎

def process_order(order_id):
    order = db.get_order(order_id)
    inventory = db.get_inventory(order.sku)
    if inventory.quantity >= order.quantity:
        db.decrement_inventory(order.sku, order.quantity)
        db.update_order_status(order_id, "confirmed")
        send_confirmation_email(order.user_email)

def process_order(order_id):
    try:
        order = db.get_order(order_id)
        if not order:
            return {"status": "error", "reason": "order_not_found"}

        inventory = db.get_inventory(order.sku)
        if not inventory:
            return {"status": "error", "reason": "sku_not_found"}

        if inventory.quantity < order.quantity:
            return {"status": "error", "reason": "insufficient_inventory"}

        # 扣库存和更新订单应该是原子的
        with db.transaction():
            db.decrement_inventory(order.sku, order.quantity)
            db.update_order_status(order_id, "confirmed")

        # 邮件发送失败不应该影响订单确认
        try:
            send_confirmation_email(order.user_email)
        except EmailException:
            log.error(f"邮件发送失败，订单 {order_id} 已确认", exc_info=True)
            # 加入重试队列，不阻塞主流程

        return {"status": "ok", "order_id": order_id}

    except DatabaseException as e:
        log.error(f"数据库异常，订单 {order_id}", exc_info=True)
        return {"status": "error", "reason": "database_error", "detail": str(e)}

# Demo 级的"错误处理"——异常被吞了，没人知道
try:
    do_something()
except:
    pass

# 生产级的错误处理——每个分支都有日志、有状态、有回退
try:
    do_something()
except TimeoutError:
    log.warning("超时，使用缓存数据")
    return get_cached_result()
except ConnectionError:
    log.error("连接失败，通知监控")
    alert_monitoring("service_down")
    raise  # 让上层决定重试还是降级

# 一个全局计数器
total_requests = 0

def handle_request():
    global total_requests
    total_requests += 1
    # ... 处理逻辑 ...

import threading

_lock = threading.Lock()
total_requests = 0

def handle_request():
    with _lock:
        global total_requests
        total_requests += 1
    # 锁外做业务逻辑，锁内只做数据修改
    process()

# ❌ AI 给的：读和写不在同一保护域
config = json.load(open("config.json"))  # 读
# 另一个线程可能已经写入了
config["last_update"] = now()             # 写（基于过期数据）
json.dump(config, open("config.json", "w"))

# ✅ 生产级：读改写原子化
with _config_lock:
    config = json.load(open("config.json"))
    config["last_update"] = now()
    json.dump(config, open("config.json", "w"))

def read_large_file(path):
    data = open(path).read()
    return process(data)

# ✅ 上下文管理器是唯一安全的写法
def read_large_file(path):
    with open(path) as f:
        data = f.read()
    return process(data)

# ✅ 不只文件——连接池、线程池、锁都需要生命周期管理
class DatabasePool:
    def __init__(self, max_connections=10):
        self.pool = Queue(maxsize=max_connections)

    def __enter__(self):
        for _ in range(self.pool.maxsize):
            self.pool.put(create_connection())
        return self

    def __exit__(self, *args):
        while not self.pool.empty():
            conn = self.pool.get()
            conn.close()

def process_payment(amount, user_id):
    result = payment_gateway.charge(amount, user_id)
    return result

import logging

logger = logging.getLogger(__name__)

def process_payment(amount, user_id):
    logger.info(f"开始处理支付: user={user_id}, amount={amount}")

    start = time.time()
    try:
        result = payment_gateway.charge(amount, user_id, timeout=30)
        elapsed = time.time() - start

        logger.info(
            f"支付成功: user={user_id}, amount={amount}, "
            f"耗时={elapsed:.2f}s, txn_id={result.transaction_id}"
        )
        return result

    except TimeoutError:
        elapsed = time.time() - start
        logger.error(
            f"支付超时: user={user_id}, amount={amount}, "
            f"耗时={elapsed:.2f}s（超过30s阈值）"
        )
        return {"status": "timeout", "user_id": user_id}

    except PaymentDeclinedError as e:
        logger.warning(
            f"支付被拒绝: user={user_id}, reason={e.reason}, code={e.code}"
        )
        return {"status": "declined", "reason": e.reason}

def get_user_profile(user_id):
    response = requests.get(f"https://api.service.com/users/{user_id}")
    return response.json()

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def get_user_profile(user_id):
    try:
        response = requests.get(
            f"https://api.service.com/users/{user_id}",
            timeout=10
        )

        if response.status_code == 404:
            return None  # 用户不存在，不重试

        if response.status_code == 429:
            # 被限流，等更久再重试
            raise RateLimitError()

        response.raise_for_status()
        return response.json()

    except requests.Timeout:
        logger.warning(f"获取用户信息超时: user={user_id}")
        return get_cached_profile(user_id)  # 回退到缓存

    except requests.ConnectionError:
        logger.error(f"服务不可达: user={user_id}")
        raise  # 让上层决定是否熔断

# 1. 超时
timeout=10  # 永远不要默认无超时

# 2. 重试（带退避）
@retry(stop=stop_after_attempt(3), wait=wait_exponential())

# 3. 熔断
circuit_breaker.record_failure()
if circuit_breaker.is_open():
    return fallback()

# 4. 降级
try:
    result = primary_service.call()
except:
    result = fallback_service.call()

# 5. 限流
rate_limiter.acquire()  # 保护自己也保护对方

[ ] 每个异常分支有明确的业务含义？（不是 bare except）
[ ] 共享状态有并发保护？（锁覆盖了读+改+写全流程）
[ ] 文件/连接/锁都有明确的生命周期？（with / __exit__）
[ ] 关键路径有结构化日志？（包含上下文、耗时、结果）
[ ] 外部调用有超时+重试+降级？（不是假设对方永远正常）