抽象层级的艺术：从机器码到自然语言

# Python：接近伪代码的表达
class Calculator:
    def __init__(self):
        self.history = []  # 动态类型，无需声明
    
    def add(self, a, b):
        result = a + b
        self.history.append(f"{a} + {b} = {result}")
        return result
    
    def get_history(self):
        return self.history.copy()

# 使用如此简单
calc = Calculator()
print(calc.add(5, 3))      # 8
print(calc.add(10, 20))    # 30
print(calc.get_history())  # ['5 + 3 = 8', '10 + 20 = 30']

# 函数式风格：声明式而非命令式
from functools import reduce

numbers = [1, 2, 3, 4, 5]

# 命令式：告诉计算机如何计算
sum_result = 0
for n in numbers:
    sum_result += n

# 声明式：描述期望的结果
sum_result = sum(numbers)  # 或 reduce(lambda a, b: a + b, numbers)

# 更复杂的管道操作
even_squares_sum = sum(x**2 for x in numbers if x % 2 == 0)
# 读作："偶数的平方和" —— 几乎就是自然语言

# Python实现：1行
open('errors.log', 'w').writelines(line for line in open('input.log') if 'ERROR' in line)

# 或者更具可读性：
with open('input.log') as infile, open('errors.log', 'w') as outfile:
    outfile.writelines(line for line in infile if 'ERROR' in line)

# AI生成的完整解决方案
import os
import time
import logging
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import boto3
from botocore.exceptions import ClientError
from slack_sdk.webhook import WebhookClient
from tenacity import retry, stop_after_attempt, wait_exponential

# 配置
WATCH_DIR = os.getenv('WATCH_DIR', '/path/to/watch')
S3_BUCKET = os.getenv('S3_BUCKET', 'my-backup-bucket')
SLACK_WEBHOOK = os.getenv('SLACK_WEBHOOK_URL')

# 日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class S3Uploader(FileSystemEventHandler):
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.slack = WebhookClient(SLACK_WEBHOOK) if SLACK_WEBHOOK else None
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    def upload_to_s3(self, file_path: str):
        """上传文件到S3，带重试机制"""
        key = Path(file_path).name
        try:
            self.s3.upload_file(file_path, S3_BUCKET, key)
            logger.info(f"✓ 上传成功: {key}")
            return key
        except ClientError as e:
            logger.error(f"✗ 上传失败: {e}")
            raise
    
    def send_notification(self, filename: str, success: bool):
        """发送Slack通知"""
        if not self.slack:
            return
        
        emoji = "✅" if success else "❌"
        text = f"{emoji} 文件 `{filename}` {'上传成功' if success else '上传失败'}"
        
        try:
            response = self.slack.send(text=text)
            if response.status_code != 200:
                logger.warning(f"Slack通知发送失败: {response.body}")
        except Exception as e:
            logger.error(f"Slack通知异常: {e}")
    
    def on_created(self, event):
        """当有新文件创建时触发"""
        if event.is_directory:
            return
        
        file_path = event.src_path
        filename = Path(file_path).name
        
        logger.info(f"检测到新文件: {filename}")
        
        # 等待文件写入完成（处理大文件）
        time.sleep(1)
        
        try:
            key = self.upload_to_s3(file_path)
            self.send_notification(filename, success=True)
        except Exception as e:
            logger.error(f"处理文件失败: {e}")
            self.send_notification(filename, success=False)

def main():
    event_handler = S3Uploader()
    observer = Observer()
    observer.schedule(event_handler, WATCH_DIR, recursive=False)
    observer.start()
    
    logger.info(f"👁️ 开始监控目录: {WATCH_DIR}")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        logger.info("👋 监控已停止")
    
    observer.join()

if __name__ == "__main__":
    main()

# Python让你感觉整数没有限制
x = 2 ** 1000  # 完全没问题！

# 但当你做大量计算时...
import time

# 小整数
start = time.time()
for i in range(1000000):
    _ = i + 1
print(f"小整数: {time.time() - start:.3f}s")

# 大整数  
start = time.time()
big = 2 ** 1000
for i in range(1000000):
    _ = big + 1
print(f"大整数: {time.time() - start:.3f}s")
# 大整数运算慢得多，因为底层需要动态内存分配

# 一个"简单"的Python程序
from sqlalchemy import create_engine

def get_user(user_id):
    engine = create_engine('postgresql://...')
    with engine.connect() as conn:
        result = conn.execute("SELECT * FROM users WHERE id = %s", user_id)
        return result.fetchone()

# nlp_interface.py
from langchain import OpenAI, LLMChain, PromptTemplate

QUERY_TEMPLATE = """
将用户的自然语言问题转换为结构化日志查询。

可用字段：
- service: 服务名称 (payment, auth, inventory, ...)
- level: 日志级别 (DEBUG, INFO, WARN, ERROR, FATAL)
- timestamp: 时间戳
- message: 日志内容
- trace_id: 分布式追踪ID

用户问题：{question}

以JSON格式返回查询参数：
"""

def natural_language_to_query(user_question: str) -> dict:
    llm = OpenAI(temperature=0)
    chain = LLMChain(llm=llm, prompt=PromptTemplate(
        template=QUERY_TEMPLATE,
        input_variables=["question"]
    ))
    
    result = chain.run(question=user_question)
    return json.loads(result)

# 使用
query = natural_language_to_query("过去一小时支付服务有什么异常？")
# 返回: {"service": "payment", "time_range": "1h", "level": "ERROR"}

# query_dsl.py
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta

@dataclass
class LogQuery:
    service: Optional[str] = None
    level: Optional[str] = None
    time_start: Optional[datetime] = None
    time_end: Optional[datetime] = None
    pattern: Optional[str] = None
    
    def service(self, name: str) -> 'LogQuery':
        self.service = name
        return self
    
    def time_range(self, duration: str) -> 'LogQuery':
        """支持 "1h", "30m", "1d" 等语法"""
        now = datetime.now()
        self.time_end = now
        
        if duration.endswith('h'):
            self.time_start = now - timedelta(hours=int(duration[:-1]))
        elif duration.endswith('m'):
            self.time_start = now - timedelta(minutes=int(duration[:-1]))
        
        return self
    
    def to_flink_sql(self) -> str:
        """转换为Flink SQL"""
        conditions = []
        if self.service:
            conditions.append(f"service = '{self.service}'")
        if self.level:
            conditions.append(f"level = '{self.level}'")
        if self.time_start:
            conditions.append(f"timestamp >= '{self.time_start.isoformat()}'")
        
        where_clause = " AND ".join(conditions) if conditions else "1=1"
        
        return f"""
        SELECT 
            window_start,
            window_end,
            level,
            COUNT(*) as count,
            COLLECT(message) as sample_messages
        FROM TABLE(TUMBLE(TABLE logs, DESCRIPTOR(timestamp), INTERVAL '5' MINUTES))
        WHERE {where_clause}
        GROUP BY window_start, window_end, level
        """