[用户增长] → [服务器负载] → [响应时间] → [用户体验] → [用户增长]
                ↑                                    ↓
                └────── [自动伸缩] ←────────────┘

# ADR-001: 采用 PostgreSQL 作为主数据库

## 状态
已接受

## 背景
- 需要 ACID 事务保证
- 数据结构以关系型为主
- 团队熟悉 SQL

## 决策
选择 PostgreSQL 14

## 后果
+ 简化一致性管理
+ 团队学习成本低
- 水平扩展有限

"""
架构决策记录（ADR）管理系统

帮助架构师记录、查询和分析架构决策
"""

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
import json
import os


class ADRStatus(Enum):
    """ADR 状态"""
    PROPOSED = "proposed"       # 提议中
    ACCEPTED = "accepted"       # 已接受
    DEPRECATED = "deprecated"   # 已废弃
    SUPERSEDED = "superseded"   # 已被替代


class ImpactLevel(Enum):
    """影响级别"""
    LOW = "low"           # 局部影响
    MEDIUM = "medium"     # 团队级影响
    HIGH = "high"         # 跨团队影响
    CRITICAL = "critical" # 组织级影响


@dataclass
class Stakeholder:
    """利益相关者"""
    name: str
    role: str
    attitude: str = "neutral"  # supportive, neutral, opposed
    
    def to_dict(self):
        return {
            "name": self.name,
            "role": self.role,
            "attitude": self.attitude
        }
    
    @classmethod
    def from_dict(cls, data: dict):
        return cls(**data)


@dataclass
class ArchitectureDecision:
    """架构决策记录"""
    
    title: str
    status: ADRStatus
    context: str
    decision: str
    consequences: List[str]
    alternatives: List[Dict[str, str]] = field(default_factory=list)
    stakeholders: List[Stakeholder] = field(default_factory=list)
    impact_level: ImpactLevel = ImpactLevel.MEDIUM
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    references: List[str] = field(default_factory=list)
    
    def add_stakeholder(self, stakeholder: Stakeholder):
        """添加利益相关者"""
        self.stakeholders.append(stakeholder)
    
    def add_alternative(self, name: str, description: str, pros: List[str], cons: List[str]):
        """添加备选方案"""
        self.alternatives.append({
            "name": name,
            "description": description,
            "pros": pros,
            "cons": cons
        })
    
    def add_tag(self, tag: str):
        """添加标签"""
        if tag not in self.tags:
            self.tags.append(tag)
    
    def update_status(self, new_status: ADRStatus):
        """更新状态"""
        old_status = self.status
        self.status = new_status
        self.updated_at = datetime.now().isoformat()
        return f"Status changed from {old_status.value} to {new_status.value}"
    
    def to_dict(self) -> dict:
        """转换为字典"""
        return {
            "title": self.title,
            "status": self.status.value,
            "context": self.context,
            "decision": self.decision,
            "consequences": self.consequences,
            "alternatives": self.alternatives,
            "stakeholders": [s.to_dict() for s in self.stakeholders],
            "impact_level": self.impact_level.value,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "tags": self.tags,
            "references": self.references
        }
    
    def to_markdown(self) -> str:
        """转换为 Markdown 格式"""
        lines = [
            f"# {self.title}",
            "",
            f"**状态**: {self.status.value}",
            f"**影响级别**: {self.impact_level.value}",
            f"**创建时间**: {self.created_at}",
            "",
            "## 背景",
            self.context,
            "",
            "## 决策",
            self.decision,
            "",
            "## 后果",
        ]
        
        for consequence in self.consequences:
            prefix = "+" if consequence.startswith("+") else "-"
            lines.append(f"- {prefix} {consequence.lstrip('+- ')}")
        
        if self.alternatives:
            lines.extend(["", "## 考虑的替代方案"])
            for alt in self.alternatives:
                lines.append(f"\n### {alt['name']}")
                lines.append(alt['description'])
                lines.append("\n**优点**:")
                for pro in alt['pros']:
                    lines.append(f"- {pro}")
                lines.append("\n**缺点**:")
                for con in alt['cons']:
                    lines.append(f"- {con}")
        
        if self.stakeholders:
            lines.extend(["", "## 利益相关者"])
            for s in self.stakeholders:
                lines.append(f"- {s.name} ({s.role}): {s.attitude}")
        
        if self.tags:
            lines.extend(["", f"**标签**: {', '.join(self.tags)}"])
        
        if self.references:
            lines.extend(["", "## 参考"])
            for ref in self.references:
                lines.append(f"- {ref}")
        
        return "\n".join(lines)
    
    @classmethod
    def from_dict(cls, data: dict) -> 'ArchitectureDecision':
        """从字典创建"""
        stakeholders = [Stakeholder.from_dict(s) for s in data.get("stakeholders", [])]
        return cls(
            title=data["title"],
            status=ADRStatus(data["status"]),
            context=data["context"],
            decision=data["decision"],
            consequences=data["consequences"],
            alternatives=data.get("alternatives", []),
            stakeholders=stakeholders,
            impact_level=ImpactLevel(data.get("impact_level", "medium")),
            created_at=data.get("created_at"),
            updated_at=data.get("updated_at"),
            tags=data.get("tags", []),
            references=data.get("references", [])
        )


class ADRRegistry:
    """ADR 注册中心"""
    
    def __init__(self, storage_path: str = "./adrs"):
        self.storage_path = storage_path
        self._adrs: Dict[int, ArchitectureDecision] = {}
        self._next_id = 1
        self._load_existing()
    
    def _load_existing(self):
        """加载已有的 ADR"""
        if not os.path.exists(self.storage_path):
            os.makedirs(self.storage_path)
            return
        
        for filename in os.listdir(self.storage_path):
            if filename.endswith(".json"):
                filepath = os.path.join(self.storage_path, filename)
                with open(filepath, 'r') as f:
                    data = json.load(f)
                    adr_id = int(filename.replace(".json", "").split("-")[0])
                    self._adrs[adr_id] = ArchitectureDecision.from_dict(data)
                    if adr_id >= self._next_id:
                        self._next_id = adr_id + 1
    
    def register(self, adr: ArchitectureDecision) -> int:
        """注册新的 ADR"""
        adr_id = self._next_id
        self._adrs[adr_id] = adr
        self._next_id += 1
        self._save(adr_id, adr)
        return adr_id
    
    def get(self, adr_id: int) -> Optional[ArchitectureDecision]:
        """获取 ADR"""
        return self._adrs.get(adr_id)
    
    def update(self, adr_id: int, adr: ArchitectureDecision):
        """更新 ADR"""
        self._adrs[adr_id] = adr
        self._save(adr_id, adr)
    
    def list_by_status(self, status: ADRStatus) -> List[tuple]:
        """按状态列出 ADR"""
        return [
            (id, adr.title) 
            for id, adr in self._adrs.items() 
            if adr.status == status
        ]
    
    def list_by_tag(self, tag: str) -> List[tuple]:
        """按标签列出 ADR"""
        return [
            (id, adr.title) 
            for id, adr in self._adrs.items() 
            if tag in adr.tags
        ]
    
    def search(self, keyword: str) -> List[tuple]:
        """搜索 ADR"""
        results = []
        for id, adr in self._adrs.items():
            if (keyword.lower() in adr.title.lower() or
                keyword.lower() in adr.context.lower() or
                keyword.lower() in adr.decision.lower()):
                results.append((id, adr.title))
        return results
    
    def export_all(self) -> List[dict]:
        """导出所有 ADR"""
        return [
            {"id": id, **adr.to_dict()} 
            for id, adr in self._adrs.items()
        ]
    
    def _save(self, adr_id: int, adr: ArchitectureDecision):
        """保存 ADR 到文件"""
        filepath = os.path.join(self.storage_path, f"{adr_id}.json")
        with open(filepath, 'w') as f:
            json.dump(adr.to_dict(), f, indent=2, ensure_ascii=False)
    
    def generate_report(self) -> str:
        """生成 ADR 报告"""
        total = len(self._adrs)
        by_status = {}
        for adr in self._adrs.values():
            status = adr.status.value
            by_status[status] = by_status.get(status, 0) + 1
        
        lines = [
            "# ADR 报告",
            "",
            f"总决策数: {total}",
            "",
            "## 按状态分布",
        ]
        
        for status, count in sorted(by_status.items()):
            lines.append(f"- {status}: {count}")
        
        lines.extend(["", "## 最近决策"])
        
        recent = sorted(
            self._adrs.items(), 
            key=lambda x: x[1].created_at, 
            reverse=True
        )[:5]
        
        for id, adr in recent:
            lines.append(f"- [{id}] {adr.title} ({adr.status.value})")
        
        return "\n".join(lines)


# ==================== 使用示例 ====================

def demo_adr_system():
    """演示 ADR 系统"""
    print("\n===== ADR 系统演示 =====")
    
    registry = ADRRegistry()
    
    # 创建一个 ADR
    database_adr = ArchitectureDecision(
        title="采用 PostgreSQL 作为主数据库",
        status=ADRStatus.ACCEPTED,
        context="""
系统需要强一致性的事务支持。
数据结构相对固定，以关系型数据为主。
团队熟悉 SQL 技术栈。
预计数据量在 TB 级别，但单表不超过千万行。
        """.strip(),
        decision="""
选择 PostgreSQL 14 作为主数据库。

理由：
1. ACID 事务保证
2. 成熟的 JSONB 支持，兼顾灵活性
3. 活跃的社区和丰富的生态
4. 良好的性能和可扩展性
5. 团队已有使用经验
        """.strip(),
        consequences=[
            "+ 简化了数据一致性管理",
            "+ 团队学习成本低",
            "+ JSONB 支持半结构化数据",
            "- 水平扩展能力有限，未来可能需要分库分表",
            "- 相比 NoSQL，写入性能略低"
        ],
        impact_level=ImpactLevel.HIGH,
        tags=["database", "infrastructure"]
    )
    
    # 添加备选方案
    database_adr.add_alternative(
        name="MongoDB",
        description="NoSQL 文档数据库",
        pros=["水平扩展能力强", "灵活的 schema", "高写入吞吐"],
        cons=["最终一致性", "JOIN 操作复杂", "团队经验不足"]
    )
    
    database_adr.add_alternative(
        name="MySQL",
        description="传统关系型数据库",
        pros=["广泛使用", "成熟稳定", "资源丰富"],
        cons=["JSON 支持不如 PG", "高级功能较少"]
    )
    
    # 添加利益相关者
    database_adr.add_stakeholder(Stakeholder("张三", "Tech Lead", "supportive"))
    database_adr.add_stakeholder(Stakeholder("李四", "DBA", "neutral"))
    database_adr.add_stakeholder(Stakeholder("王五", "产品经理", "neutral"))
    
    # 注册
    adr_id = registry.register(database_adr)
    print(f"ADR 已注册，ID: {adr_id}")
    
    # 创建另一个 ADR
    api_adr = ArchitectureDecision(
        title="采用 RESTful API 作为主要通信方式",
        status=ADRStatus.ACCEPTED,
        context="""
系统需要与多个外部系统集成。
团队成员熟悉 REST 范式。
客户端包括 Web、移动端和第三方合作伙伴。
        """.strip(),
        decision="""
选择 RESTful API 作为主要通信方式，基于 HTTP/JSON。
使用 OpenAPI 规范定义接口契约。
        """.strip(),
        consequences=[
            "+ 广泛的技术支持和工具链",
            "+ 易于理解和调试",
            "+ 良好的缓存支持",
            "- 相比 gRPC，序列化效率较低",
            "- 实时双向通信需要额外方案"
        ],
        impact_level=ImpactLevel.MEDIUM,
        tags=["api", "integration"]
    )
    
    registry.register(api_adr)
    
    # 查询和输出
    print("\n" + "=" * 60)
    print("所有已接受的架构决策:")
    print("=" * 60)
    
    for id, title in registry.list_by_status(ADRStatus.ACCEPTED):
        print(f"[{id}] {title}")
    
    # 查看详细信息
    print("\n" + "=" * 60)
    print("数据库决策详情:")
    print("=" * 60)
    
    adr = registry.get(1)
    if adr:
        print(adr.to_markdown())
    
    # 搜索
    print("\n" + "=" * 60)
    print("搜索 '数据库':")
    print("=" * 60)
    
    results = registry.search("数据库")
    for id, title in results:
        print(f"[{id}] {title}")
    
    # 生成报告
    print("\n" + "=" * 60)
    print(registry.generate_report())


if __name__ == "__main__":
    demo_adr_system()

"""
系统思维练习工具

帮助架构师可视化和分析系统中的因果关系和反馈回路
"""

from dataclasses import dataclass, field
from typing import List, Dict, Set, Tuple
from enum import Enum
import json


class RelationshipType(Enum):
    """关系类型"""
    POSITIVE = "positive"   # 正相关（同向变化）
    NEGATIVE = "negative"   # 负相关（反向变化）


@dataclass
class Variable:
    """系统变量"""
    name: str
    description: str
    current_value: float = 0.0  # 归一化值 -1 到 1
    
    def to_dict(self):
        return {
            "name": self.name,
            "description": self.description,
            "current_value": self.current_value
        }


@dataclass
class Relationship:
    """因果关系"""
    source: str
    target: str
    type: RelationshipType
    strength: float = 1.0  # 强度 0-1
    delay: float = 0.0     # 延迟（时间单位）
    
    def describe(self) -> str:
        direction = "increases" if self.type == RelationshipType.POSITIVE else "decreases"
        return f"{self.source} {direction} {self.target} (strength: {self.strength}, delay: {self.delay})"


@dataclass
class FeedbackLoop:
    """反馈回路"""
    variables: List[str]
    type: str  # "reinforcing" (positive) or "balancing" (negative)
    description: str
    
    def is_reinforcing(self) -> bool:
        return self.type == "reinforcing"
    
    def is_balancing(self) -> bool:
        return self.type == "balancing"


class CausalLoopDiagram:
    """因果回路图"""
    
    def __init__(self, name: str):
        self.name = name
        self.variables: Dict[str, Variable] = {}
        self.relationships: List[Relationship] = []
        self.loops: List[FeedbackLoop] = []
    
    def add_variable(self, name: str, description: str, initial_value: float = 0.0):
        """添加变量"""
        self.variables[name] = Variable(name, description, initial_value)
    
    def add_relationship(self, source: str, target: str, 
                        rel_type: RelationshipType, strength: float = 1.0, delay: float = 0.0):
        """添加因果关系"""
        if source not in self.variables:
            raise ValueError(f"Unknown variable: {source}")
        if target not in self.variables:
            raise ValueError(f"Unknown variable: {target}")
        
        self.relationships.append(Relationship(source, target, rel_type, strength, delay))
    
    def identify_loops(self) -> List[FeedbackLoop]:
        """识别反馈回路"""
        loops = []
        visited = set()
        
        # 构建邻接表
        graph: Dict[str, List[Tuple[str, RelationshipType]]] = {}
        for var_name in self.variables:
            graph[var_name] = []
        
        for rel in self.relationships:
            graph[rel.source].append((rel.target, rel.type))
        
        # DFS 查找回路
        def find_loops_from(start: str, current: str, path: List[str], 
                           visited_edges: Set[Tuple[str, str]]):
            for neighbor, rel_type in graph[current]:
                edge = (current, neighbor)
                if edge in visited_edges:
                    continue
                
                if neighbor == start and len(path) > 2:
                    # 找到回路
                    loop_vars = path.copy()
                    loop_type = self._determine_loop_type(path, graph)
                    loops.append(FeedbackLoop(
                        variables=loop_vars,
                        type=loop_type,
                        description=self._describe_loop(loop_vars)
                    ))
                elif neighbor not in path:
                    new_visited = visited_edges | {edge}
                    find_loops_from(start, neighbor, path + [neighbor], new_visited)
        
        for var_name in self.variables:
            find_loops_from(var_name, var_name, [var_name], set())
        
        # 去重
        unique_loops = []
        seen = set()
        for loop in loops:
            key = tuple(sorted(loop.variables))
            if key not in seen:
                seen.add(key)
                unique_loops.append(loop)
        
        self.loops = unique_loops
        return unique_loops
    
    def _determine_loop_type(self, path: List[str], 
                            graph: Dict[str, List[Tuple[str, RelationshipType]]]) -> str:
        """确定回路类型"""
        negative_count = 0
        
        for i in range(len(path)):
            source = path[i]
            target = path[(i + 1) % len(path)]
            
            for neighbor, rel_type in graph[source]:
                if neighbor == target:
                    if rel_type == RelationshipType.NEGATIVE:
                        negative_count += 1
                    break
        
        # 奇数个负关系 = 平衡回路，偶数个 = 增强回路
        return "balancing" if negative_count % 2 == 1 else "reinforcing"
    
    def _describe_loop(self, variables: List[str]) -> str:
        """描述回路"""
        var_names = " → ".join(variables)
        return f"Loop: {var_names}"
    
    def analyze(self) -> Dict:
        """分析系统"""
        loops = self.identify_loops()
        
        reinforcing = [l for l in loops if l.is_reinforcing()]
        balancing = [l for l in loops if l.is_balancing()]
        
        analysis = {
            "name": self.name,
            "variable_count": len(self.variables),
            "relationship_count": len(self.relationships),
            "loop_count": len(loops),
            "reinforcing_loops": len(reinforcing),
            "balancing_loops": len(balancing),
            "loops": [
                {
                    "variables": loop.variables,
                    "type": loop.type,
                    "description": loop.description
                }
                for loop in loops
            ],
            "relationships": [rel.describe() for rel in self.relationships]
        }
        
        return analysis
    
    def to_dict(self) -> dict:
        """转换为字典"""
        return {
            "name": self.name,
            "variables": {k: v.to_dict() for k, v in self.variables.items()},
            "relationships": [
                {
                    "source": r.source,
                    "target": r.target,
                    "type": r.type.value,
                    "strength": r.strength,
                    "delay": r.delay
                }
                for r in self.relationships
            ],
            "loops": [
                {
                    "variables": l.variables,
                    "type": l.type,
                    "description": l.description
                }
                for l in self.loops
            ]
        }
    
    def visualize_text(self) -> str:
        """文本可视化"""
        lines = [f"=== {self.name} ===", ""]
        
        lines.append("变量:")
        for name, var in self.variables.items():
            lines.append(f"  - {name}: {var.description}")
        
        lines.append("\n因果关系:")
        for rel in self.relationships:
            arrow = "→" if rel.type == RelationshipType.POSITIVE else "↘"
            lines.append(f"  {rel.source} {arrow} {rel.target} (strength: {rel.strength})")
        
        if self.loops:
            lines.append("\n反馈回路:")
            for loop in self.loops:
                loop_type = "🔄 增强" if loop.is_reinforcing() else "⚖️ 平衡"
                lines.append(f"  {loop_type}: {' → '.join(loop.variables)}")
        
        return "\n".join(lines)


# ==================== 示例：电商系统分析 ====================

def demo_ecommerce_system():
    """演示电商系统的因果回路分析"""
    print("\n===== 电商系统因果回路分析 =====")
    
    # 创建因果回路图
    cl = CausalLoopDiagram("E-commerce System")
    
    # 添加变量
    cl.add_variable("marketing_spend", "营销投入")
    cl.add_variable("user_acquisition", "用户获取")
    cl.add_variable("active_users", "活跃用户数")
    cl.add_variable("revenue", "收入")
    cl.add_variable("profit", "利润")
    cl.add_variable("customer_satisfaction", "客户满意度")
    cl.add_variable("word_of_mouth", "口碑传播")
    cl.add_variable("server_load", "服务器负载")
    cl.add_variable("response_time", "响应时间")
    cl.add_variable("user_experience", "用户体验")
    cl.add_variable("infrastructure_cost", "基础设施成本")
    
    # 添加关系
    # 营销 → 用户获取 → 活跃用户 → 收入 → 利润 → 营销（增强回路）
    cl.add_relationship("marketing_spend", "user_acquisition", RelationshipType.POSITIVE, 0.8)
    cl.add_relationship("user_acquisition", "active_users", RelationshipType.POSITIVE, 0.9)
    cl.add_relationship("active_users", "revenue", RelationshipType.POSITIVE, 0.85)
    cl.add_relationship("revenue", "profit", RelationshipType.POSITIVE, 0.7)
    cl.add_relationship("profit", "marketing_spend", RelationshipType.POSITIVE, 0.6)
    
    # 活跃用户 → 口碑 → 用户获取（增强回路）
    cl.add_relationship("active_users", "customer_satisfaction", RelationshipType.POSITIVE, 0.5)
    cl.add_relationship("customer_satisfaction", "word_of_mouth", RelationshipType.POSITIVE, 0.7)
    cl.add_relationship("word_of_mouth", "user_acquisition", RelationshipType.POSITIVE, 0.6)
    
    # 活跃用户 → 服务器负载 → 响应时间 → 用户体验 → 活跃度（平衡回路）
    cl.add_relationship("active_users", "server_load", RelationshipType.POSITIVE, 0.9)
    cl.add_relationship("server_load", "response_time", RelationshipType.POSITIVE, 0.8)
    cl.add_relationship("response_time", "user_experience", RelationshipType.NEGATIVE, 0.85)
    cl.add_relationship("user_experience", "active_users", RelationshipType.POSITIVE, 0.7)
    
    # 收入 → 基础设施成本 → 利润（平衡回路）
    cl.add_relationship("revenue", "infrastructure_cost", RelationshipType.POSITIVE, 0.5)
    cl.add_relationship("infrastructure_cost", "profit", RelationshipType.NEGATIVE, 0.6)
    
    # 分析
    analysis = cl.analyze()
    
    print(cl.visualize_text())
    
    print("\n" + "=" * 60)
    print("分析结果:")
    print("=" * 60)
    print(f"变量数: {analysis['variable_count']}")
    print(f"关系数: {analysis['relationship_count']}")
    print(f"反馈回路数: {analysis['loop_count']}")
    print(f"  - 增强回路: {analysis['reinforcing_loops']}")
    print(f"  - 平衡回路: {analysis['balancing_loops']}")
    
    print("\n详细回路:")
    for loop in analysis['loops']:
        loop_type = "增强" if loop['type'] == 'reinforcing' else "平衡"
        print(f"  [{loop_type}] {' → '.join(loop['variables'])}")


if __name__ == "__main__":
    demo_ecommerce_system()