from pygaeb import GAEBParser

doc = GAEBParser.parse("tender.X83")   # DA XML 3.x
doc = GAEBParser.parse("old.D83")     # DA XML 2.x — same call

print(doc.source_version)   # e.g. SourceVersion.DA_XML_33
print(doc.exchange_phase)   # ExchangePhase.X83
print(doc.item_count)       # Total number of items
print(doc.grand_total)      # Decimal("1234567.89")

for item in doc.iter_items():
    print(item.oz)           # "01.02.0030"
    print(item.short_text)   # "Mauerwerk der Innenwand…"
    print(item.qty)          # Decimal("1170.000")
    print(item.unit)         # "m2"
    print(item.unit_price)   # Decimal("45.50")
    print(item.total_price) # Decimal("53235.00")
    print(item.item_type)    # ItemType.NORMAL

# Document summary
print(doc.item_count)           # Total item count
print(doc.document_kind)        # e.g. DocumentKind.PROCUREMENT, TRADE, COST, QUANTITY
print(doc.gaeb_info.version)   # GAEB software version that created the file

# Project info (AwardInfo)
a = doc.award
print(a.project_name, a.project_no, a.description)
print(a.open_date, a.construction_start, a.construction_end)
print(a.currency_label, a.contract_no)
print(a.warranty_duration, a.warranty_unit)
print(a.bid_comm_perm, a.alter_bid_perm)   # Bid permissions

# Owner / client (OWN)
print(a.client, a.award_no)
addr = a.owner_address
if addr:
    print(addr.name, addr.street, addr.pcode, addr.city, addr.country, addr.contact)

boq = doc.award.boq
print(boq.is_multi_lot)   # True if document has multiple lots

for lot in boq.lots:
    print(lot.label or lot.rno)           # Lot label / number
    for ctgy in lot.body.categories:
        print("  ", ctgy.rno, ctgy.label)  # Category number and title
        for item in ctgy.items:
            print("    ", item.oz, item.short_text, item.qty, item.unit)
        if ctgy.totals and ctgy.totals.total is not None:
            print("    Subtotal:", ctgy.totals.total)
        for sub in ctgy.subcategories:     # Nested subcategories
            # ... same pattern for sub.rno, sub.label, sub.items

from pygaeb import GAEBParser, ValidationMode

doc = GAEBParser.parse("tender.X83")
for issue in doc.validation_results:
    print(issue.severity, issue.message)

# For CI or strict pipelines:
doc = GAEBParser.parse("tender.X83", validation=ValidationMode.STRICT)

from pygaeb import GAEBParser, GAEBWriter, ExchangePhase
from decimal import Decimal

doc = GAEBParser.parse("tender.X83")
item = doc.award.boq.get_item("01.02.0030")
item.unit_price = Decimal("48.00")

GAEBWriter.write(doc, "bid.X84", phase=ExchangePhase.X84)

from pygaeb import GAEBConverter, SourceVersion

# Upgrade 2.x → 3.3
report = GAEBConverter.convert("old.D83", "modern.X83")

# Downgrade for compatibility
report = GAEBConverter.convert(
    "tender.X83", "compat.X83",
    target_version=SourceVersion.DA_XML_32,
)
print(f"Converted {report.items_converted} items, data loss: {report.has_data_loss}")

from pygaeb.convert import to_json, to_csv

to_json(doc, "boq.json")   # Full nested BoQ tree
to_csv(doc, "items.csv")   # Flat item table (optionally with classification columns)

from pygaeb import LLMClassifier

classifier = LLMClassifier(model="anthropic/claude-sonnet-4-6")
# classifier = LLMClassifier(model="gpt-4o")
# classifier = LLMClassifier(model="ollama/llama3")  # Local, no API key

# Check cost before running
estimate = await classifier.estimate_cost(doc)
print(f"Will classify {estimate.items_to_classify} items for ~${estimate.estimated_cost_usd:.2f}")

await classifier.enrich(doc)

for item in doc.iter_items():
    if item.classification:
        print(item.oz, item.classification.element_type, item.classification.confidence)

from pydantic import BaseModel, Field
from typing import Optional
from pygaeb import StructuredExtractor

class DoorSpec(BaseModel):
    door_type: str = Field("", description="single, double, sliding")
    width_mm: Optional[int] = Field(None, description="Width in mm")
    fire_rating: Optional[str] = Field(None, description="T30, T60, T90")
    glazing: bool = Field(False, description="Has glass panels")
    material: str = Field("", description="wood, steel, aluminium")

extractor = StructuredExtractor(model="anthropic/claude-sonnet-4-6")
doors = await extractor.extract(doc, schema=DoorSpec, element_type="Door")
for item, spec in doors:
    print(item.oz, spec.door_type, spec.fire_rating, spec.width_mm)

# Core parser, writer, export — no LLM dependencies
pip install pyGAEB

# With LLM classification (LiteLLM + instructor)
pip install pyGAEB[llm]

sudo apt update && sudo apt install -y nginx

# /etc/nginx/sites-available/myapp.conf

upstream node_backend {
    server 127.0.0.1:3000;
    keepalive 64;
}

server {
    listen 80;
    server_name myapp.example.com;
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl http2;
    server_name myapp.example.com;

    # SSL
    ssl_certificate     /etc/letsencrypt/live/myapp.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/myapp.example.com/privkey.pem;

    # Security headers
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;

    # Proxy
    location / {
        proxy_pass http://node_backend;
        proxy_http_version 1.1;

        # WebSocket support
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        proxy_cache_bypass $http_upgrade;

        # Pass client info
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Timeouts
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }

    # Static files — serve directly, skip Node
    location /static/ {
        alias /opt/myapp/public/;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    # Health check — suppress access logs
    location /health {
        proxy_pass http://node_backend;
        access_log off;
    }
}

sudo ln -s /etc/nginx/sites-available/myapp.conf /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl reload nginx

sudo apt update && sudo apt install -y apache2
sudo a2enmod proxy proxy_http proxy_wstunnel rewrite ssl headers
sudo systemctl restart apache2

# /etc/apache2/sites-available/myapp.conf

<VirtualHost *:80>
    ServerName myapp.example.com

    RewriteEngine On
    RewriteCond %{HTTPS} off
    RewriteRule ^(.*)$ https://%{HTTP_HOST}$1 [R=301,L]
</VirtualHost>

<VirtualHost *:443>
    ServerName myapp.example.com

    SSLEngine On
    SSLCertificateFile    /etc/letsencrypt/live/myapp.example.com/fullchain.pem
    SSLCertificateKeyFile /etc/letsencrypt/live/myapp.example.com/privkey.pem

    # Security headers
    Header always set X-Frame-Options "SAMEORIGIN"
    Header always set X-Content-Type-Options "nosniff"
    Header always set X-XSS-Protection "1; mode=block"
    Header always set Strict-Transport-Security "max-age=31536000; includeSubDomains"
    Header always set Referrer-Policy "strict-origin-when-cross-origin"

    # Proxy to Node.js
    ProxyPreserveHost On
    ProxyPass / http://127.0.0.1:3000/
    ProxyPassReverse / http://127.0.0.1:3000/

    # WebSocket support
    RewriteEngine On
    RewriteCond %{HTTP:Upgrade} =websocket [NC]
    RewriteRule /(.*) ws://127.0.0.1:3000/$1 [P,L]

    # Pass real client IP
    RequestHeader set X-Real-IP "%{REMOTE_ADDR}s"
    RequestHeader set X-Forwarded-Proto "https"

    # Timeout
    ProxyTimeout 60

    # Static files
    Alias /static /opt/myapp/public
    <Directory /opt/myapp/public>
        Require all granted
        Options -Indexes
        Header set Cache-Control "public, max-age=2592000, immutable"
    </Directory>

    # Logging
    ErrorLog ${APACHE_LOG_DIR}/myapp-error.log
    CustomLog ${APACHE_LOG_DIR}/myapp-access.log combined
</VirtualHost>

sudo a2ensite myapp.conf
sudo a2dissite 000-default.conf
sudo apache2ctl configtest
sudo systemctl reload apache2

# Nginx
sudo apt install -y certbot python3-certbot-nginx
sudo certbot --nginx -d myapp.example.com

# Apache2
sudo apt install -y certbot python3-certbot-apache
sudo certbot --apache -d myapp.example.com

sudo certbot renew --dry-run

// Express
app.set('trust proxy', 1);

# PM2 running?
pm2 status

# Port listening?
ss -tlnp | grep 3000

# Reverse proxy responding?
curl -I https://myapp.example.com

# SSL valid?
echo | openssl s_client -connect myapp.example.com:443 2>/dev/null 
  | grep 'Verify return code'

# Proxy logs
sudo tail -f /var/log/nginx/access.log          # Nginx
sudo tail -f /var/log/apache2/myapp-access.log   # Apache2

Client (HTTPS:443)
    │
    ▼
┌──────────────────────┐
│  Nginx / Apache2     │  ← SSL, headers, static files
│  (port 80/443)       │
└──────────┬───────────┘
           │ proxy_pass http://127.0.0.1:3000
           ▼
┌──────────────────────┐
│  PM2 (cluster mode)  │  ← Process management, restarts
├──────────────────────┤
│  Node.js app         │  ← server.js / app.js
│  (port 3000)         │
└──────────────────────┘
           │
     NVM-managed Node binary
     ~/.nvm/versions/node/v20.18.0/bin/node

## 競合コンシューマー設計ルール

### 排他処理
- Redis Consumer GroupでWorker間の排他的メッセージ割り当て
- Consumer Groupは各メッセージを1つのWorkerにのみ配信
- 処理完了後にACK（未ACKメッセージはタイムアウト後に再割り当て）

### ワーカー管理
- ワーカー数はキュー深度に応じて自動スケール
- ヘルスチェック: 30秒ごとに生存確認
- 落ちたワーカーのPELを他のワーカーが引き継ぐ

### グレースフルシャットダウン
- SIGTERMを受信したらキュー取得を停止
- 処理中のタスクが完了するまで待機（最大30秒）
- タイムアウト後は強制終了

競合コンシューマーパターンを設計してください。

要件：
- Redis Consumer Group
- Pending Entry List（PEL）の引き継ぎ
- グレースフルシャットダウン
- 自動スケーリング

生成ファイル: src/messaging/competingConsumers/

// src/messaging/competingConsumers/consumerGroup.ts — Consumer Group Worker

export interface ConsumerGroupConfig {
  streamKey: string;      // Redis Streamキー
  groupName: string;      // Consumer Group名
  consumerName: string;   // このWorkerの識別子
  batchSize: number;      // 1回に取得するメッセージ数
  blockMs: number;        // ブロッキング読み取りのタイムアウト
  pendingTimeoutMs: number; // PELタイムアウト（この時間未ACKは再割り当て）
}

export class ConsumerGroupWorker<T> {
  private running = false;
  private activeProcessing = 0;
  private readonly maxGracefulWaitMs = 30_000;

  constructor(
    private readonly config: ConsumerGroupConfig,
    private readonly handler: (payload: T) => Promise<void>
  ) {}

  async start(): Promise<void> {
    // Consumer Groupがなければ作成
    try {
      await redis.xGroupCreate(
        this.config.streamKey,
        this.config.groupName,
        '$',  // 新規メッセージからのみ処理（既存メッセージは無視）
        { MKSTREAM: true }
      );
    } catch (error: any) {
      if (!error.message.includes('BUSYGROUP')) throw error;
      // 既に存在する場合は続行
    }

    this.running = true;
    logger.info({ consumer: this.config.consumerName, group: this.config.groupName }, 'Consumer started');

    // グレースフルシャットダウンのシグナルハンドラー
    process.on('SIGTERM', () => this.stop());
    process.on('SIGINT', () => this.stop());

    // まず未処理のPELを引き継ぐ
    await this.claimPendingMessages();

    // メインループ
    while (this.running) {
      await this.processNextBatch();
    }

    // グレースフルシャットダウン: 処理中のタスクを待つ
    await this.waitForActiveProcessing();
  }

  private async processNextBatch(): Promise<void> {
    const messages = await redis.xReadGroup(
      this.config.groupName,
      this.config.consumerName,
      { key: this.config.streamKey, id: '>' },
      { COUNT: this.config.batchSize, BLOCK: this.config.blockMs }
    );

    if (!messages || messages[0].messages.length === 0) return;

    const processingPromises = messages[0].messages.map(msg =>
      this.processMessage(msg)
    );

    await Promise.allSettled(processingPromises);
  }

  private async processMessage(msg: { id: string; message: Record<string, string> }): Promise<void> {
    this.activeProcessing++;

    try {
      const payload = JSON.parse(msg.message.payload) as T;

      await this.handler(payload);

      // 処理成功: ACK
      await redis.xAck(this.config.streamKey, this.config.groupName, msg.id);

      metrics.messagesProcessed.inc({ consumer: this.config.consumerName });
      logger.debug({ messageId: msg.id, consumer: this.config.consumerName }, 'Message acknowledged');
    } catch (error) {
      logger.error({ messageId: msg.id, consumer: this.config.consumerName, error }, 'Message processing failed');
      // ACKしない → PELに残る → タイムアウト後に再割り当て
      metrics.messagesFailed.inc({ consumer: this.config.consumerName });
    } finally {
      this.activeProcessing--;
    }
  }

  // 落ちたConsumerのPEL（未ACKメッセージ）を引き継ぐ
  private async claimPendingMessages(): Promise<void> {
    let startId = '0';

    while (true) {
      // タイムアウト超過の未ACKメッセージを自分のものにする
      const claimed = await redis.xAutoClaim(
        this.config.streamKey,
        this.config.groupName,
        this.config.consumerName,
        this.config.pendingTimeoutMs,
        startId,
        { COUNT: 100 }
      );

      if (claimed.messages.length === 0) break;

      logger.info({ count: claimed.messages.length }, 'Claimed pending messages from dead consumers');

      for (const msg of claimed.messages) {
        await this.processMessage(msg);
      }

      startId = claimed.nextId;
      if (startId === '0') break;
    }
  }

  async stop(): Promise<void> {
    logger.info({ consumer: this.config.consumerName }, 'Graceful shutdown initiated');
    this.running = false;
  }

  private async waitForActiveProcessing(): Promise<void> {
    const startMs = Date.now();

    while (this.activeProcessing > 0) {
      if (Date.now() - startMs > this.maxGracefulWaitMs) {
        logger.warn({ activeProcessing: this.activeProcessing }, 'Graceful shutdown timeout — forcing exit');
        break;
      }
      await sleep(100);
    }

    logger.info({ consumer: this.config.consumerName }, 'Consumer stopped');
  }
}

// src/messaging/competingConsumers/workerPool.ts — ワーカープール管理

export class WorkerPool<T> {
  private workers: Map<string, ConsumerGroupWorker<T>> = new Map();

  constructor(
    private readonly streamKey: string,
    private readonly groupName: string,
    private readonly handler: (payload: T) => Promise<void>
  ) {}

  async scale(targetWorkers: number): Promise<void> {
    const currentCount = this.workers.size;

    if (targetWorkers > currentCount) {
      // スケールアップ
      for (let i = currentCount; i < targetWorkers; i++) {
        const consumerName = `worker-${process.pid}-${i}`;
        const worker = new ConsumerGroupWorker<T>(
          {
            streamKey: this.streamKey,
            groupName: this.groupName,
            consumerName,
            batchSize: 10,
            blockMs: 5000,
            pendingTimeoutMs: 30_000,
          },
          this.handler
        );

        this.workers.set(consumerName, worker);
        worker.start().catch(err => logger.error({ consumerName, err }, 'Worker error'));
        logger.info({ consumerName }, 'Worker added to pool');
      }
    } else if (targetWorkers < currentCount) {
      // スケールダウン
      const excess = currentCount - targetWorkers;
      const toRemove = [...this.workers.keys()].slice(-excess);

      for (const name of toRemove) {
        const worker = this.workers.get(name);
        await worker?.stop();
        this.workers.delete(name);
        logger.info({ consumerName: name }, 'Worker removed from pool');
      }
    }
  }

  // キュー深度に応じた自動スケーリング
  async autoScale(options: { minWorkers: number; maxWorkers: number; targetLag: number }): Promise<void> {
    const info = await redis.xInfoGroups(this.streamKey);
    const group = info.find((g: any) => g.name === this.groupName);
    const lag = group?.lag ?? 0;

    let target = this.workers.size;

    if (lag > options.targetLag * 2 && this.workers.size < options.maxWorkers) {
      target = Math.min(this.workers.size + 2, options.maxWorkers);
    } else if (lag < options.targetLag / 2 && this.workers.size > options.minWorkers) {
      target = Math.max(this.workers.size - 1, options.minWorkers);
    }

    if (target !== this.workers.size) {
      logger.info({ from: this.workers.size, to: target, lag }, 'Auto-scaling worker pool');
      await this.scale(target);
    }
  }
}

// 使用例
const pool = new WorkerPool<OrderPayload>(
  'queue:order-processing',
  'order-processors',
  async (order) => {
    await processOrder(order);
  }
);

// 最初は2ワーカーで開始
await pool.scale(2);

// 30秒ごとに自動スケール（lag > 100でスケールアップ）
setInterval(() => pool.autoScale({ minWorkers: 1, maxWorkers: 10, targetLag: 100 }), 30_000);

// ❌ Using exceptions for normal outcomes
function getUser(id: string): User {
  const user = db.find(id)
  if (!user) throw new Error("User not found")
  return user
}

// ✅ Modeling errors explicitly
function getUser(id: string): User | null {
  return db.find(id) ?? null
}

// ✅ Errors are part of the function contract
func GetUser(id string) (User, error) {
    user := db.Find(id)
    if user == nil {
        return User{}, errors.New("not found")
    }
    return user, nil
}

function withdraw(amount: number) {
  if (amount < 0) {
    throw new Error("Invariant violated: negative withdrawal")
  }
}

func MustConnect() *DB {
    db, err := Connect()
    if err != nil {
        panic("cannot connect to database")
    }
    return db
}