Từ Cronjob cá nhân đến State-Machine Production: Kiến Trúc Pipeline AI Tự Trị

# pipeline/dedup.py
import re
import mmh3
from typing import List, Set

class MinHashDeduplicator:
    def __init__(self, num_perm: int = 128, threshold: float = 0.85):
        self.num_perm = num_perm
        self.threshold = threshold
        # Khởi tạo các seed ngẫu nhiên cho hàm băm MurmurHash3
        self.seeds = [i * 10007 for i in range(num_perm)]

    def _tokenize_to_shingles(self, text: str, k: int = 3) -> Set[str]:
        """Tách văn bản thành tập hợp các k-shingles (cụm 3 từ liên tiếp)"""
        words = re.findall(r'\w+', text.lower())
        if len(words) < k:
            return set(words)
        return {" ".join(words[i:i+k]) for i in range(len(words) - k + 1)}

    def compute_minhash_signature(self, text: str) -> List[int]:
        """Tính toán vector chữ ký MinHash 128 chiều bằng MurmurHash3"""
        shingles = self._tokenize_to_shingles(text)
        if not shingles:
            return [0] * self.num_perm

        signature = []
        for seed in self.seeds:
            min_val = float('inf')
            for s in shingles:
                # Băm từng shingle với seed tương ứng
                hash_val = mmh3.hash(s, seed, signed=False)
                if hash_val < min_val:
                    min_val = hash_val
            signature.append(min_val)
        return signature

    def compute_jaccard_similarity(self, sig1: List[int], sig2: List[int]) -> float:
        """Đo lường độ tương đồng Jaccard giữa hai chữ ký MinHash"""
        if len(sig1) != len(sig2):
            raise ValueError("Độ dài chữ ký MinHash không khớp")
        matching = sum(1 for a, b in zip(sig1, sig2) if a == b)
        return matching / float(self.num_perm)

    def is_duplicate(self, text: str, existing_signatures: List[List[int]]) -> bool:
        """Kiểm tra xem nội dung bài viết mới có trùng lặp với lịch sử không"""
        current_sig = self.compute_minhash_signature(text)
        for old_sig in existing_signatures:
            similarity = self.compute_jaccard_similarity(current_sig, old_sig)
            if similarity >= self.threshold:
                return True
        return False

# pipeline/quality_gate.py
import re
import subprocess
import sys
import yaml
from pathlib import Path

class ContentQualityGate:
    def __init__(self, post_path: str, hugo_root: str):
        self.post_file = Path(post_path)
        self.hugo_root = Path(hugo_root)
        self.content = self.post_file.read_text(encoding='utf-8')

    def run_all_checks(self) -> bool:
        print(f"[*] Bắt đầu kiểm định chất lượng bài viết: {self.post_file.name}")
        
        # 1. CỔNG 1: Kiểm định cấu trúc YAML Frontmatter
        if not self._check_frontmatter_schema():
            return False
            
        # 2. CỔNG 2: Phát hiện mã giả (Pseudo-code) và Placeholder cấm
        if not self._check_no_pseudocode():
            return False
            
        # 3. CỔNG 3: Kiểm định độ dài bài viết và tín hiệu E-E-A-T
        if not self._check_depth_and_eeat():
            return False
            
        # 4. CỔNG 4: Kiểm thử biên dịch thực tế với Hugo (Dry-Run Build)
        if not self._check_hugo_dry_run():
            return False

        print("[✓] Bài viết vượt qua 100% các tiêu chuẩn kiểm định chất lượng!")
        return True

    def _check_frontmatter_schema(self) -> bool:
        match = re.match(r'^---\s*\n(.*?)\n---\s*\n', self.content, re.DOTALL)
        if not match:
            print("[x] Lỗi: Không tìm thấy khối YAML Frontmatter hợp lệ.")
            return False
        try:
            meta = yaml.safe_load(match.group(1))
            required_keys = ['title', 'slug', 'date', 'author', 'canonicalURL', 'categories', 'tags']
            for k in required_keys:
                if k not in meta or not meta[k]:
                    print(f"[x] Lỗi: Thiếu trường bắt buộc '{k}' trong frontmatter.")
                    return False
            if meta.get('draft', True) is True:
                print("[x] Lỗi: Bài viết không được phép đặt 'draft: true'.")
                return False
            return True
        except yaml.YAMLError as e:
            print(f"[x] Lỗi cú pháp YAML: {e}")
            return False

    def _check_no_pseudocode(self) -> bool:
        forbidden_patterns = [
            r'//\s*(?:TODO|implement here|logic goes here)',
            r'#\s*(?:TODO|logic at here|pseudo-code)',
            r'\.\.\.\s*(?:remaining logic|other fields)',
            r'alert\(["\']implement me["\']\)'
        ]
        for pattern in forbidden_patterns:
            if re.search(pattern, self.content, re.IGNORECASE):
                print(f"[x] Lỗi: Phát hiện mã giả hoặc placeholder vi phạm: '{pattern}'")
                return False
        return True

    def _check_depth_and_eeat(self) -> bool:
        # Loại bỏ frontmatter để đếm số từ phần thân
        body = re.sub(r'^---\s*\n.*?\n---\s*\n', '', self.content, flags=re.DOTALL)
        words = len(re.findall(r'\w+', body))
        if words < 2000:
            print(f"[x] Lỗi: Bài viết quá ngắn ({words} từ). Yêu cầu tối thiểu 2.000 từ chuyên sâu.")
            return False
        if "Lê Tuấn Anh" not in self.content:
            print("[x] Lỗi: Thiếu tín hiệu quyền tác giả E-E-A-T.")
            return False
        return True

    def _check_hugo_dry_run(self) -> bool:
        cmd = ["hugo", "--renderToMemory", "-s", str(self.hugo_root)]
        res = subprocess.run(cmd, capture_output=True, text=True)
        if res.returncode != 0:
            print(f"[x] Lỗi: Hugo biên dịch thất bại:\n{res.stderr}")
            return False
        return True

if __name__ == "__main__":
    if len(sys.argv) < 3:
        sys.exit(1)
    gate = ContentQualityGate(sys.argv[1], sys.argv[2])
    if not gate.run_all_checks():
        sys.exit(1)

# pipeline/wol_orchestrator.py
import binascii
import socket
import time
import urllib.request
import paramiko

class HardwareOrchestrator:
    def __init__(self, mac_address: str, worker_ip: str, broadcast_ip: str = "192.168.1.255"):
        self.mac_address = mac_address.replace(":", "").replace("-", "")
        self.worker_ip = worker_ip
        self.broadcast_ip = broadcast_ip

    def send_magic_packet(self) -> None:
        """Gửi gói tin Magic Packet đánh thức máy chủ vật lý qua card mạng"""
        mac_bytes = binascii.unhexlify(self.mac_address)
        magic_packet = bytes([0xFF] * 6) + mac_bytes * 16
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
            s.sendto(magic_packet, (self.broadcast_ip, 9))
        print(f"[*] Đã bắn Magic Packet WOL tới thiết bị MAC {self.mac_address}")

    def wait_for_ollama_heartbeat(self, timeout_seconds: int = 120) -> bool:
        """Thăm dò cổng dịch vụ Ollama (11434) cho đến khi sẵn sàng"""
        endpoint = f"http://{self.worker_ip}:11434/api/tags"
        start_time = time.time()
        while time.time() - start_time < timeout_seconds:
            try:
                with urllib.request.urlopen(endpoint, timeout=2) as response:
                    if response.status == 200:
                        print(f"[✓] Worker Node đã sẵn sàng hoạt động sau {int(time.time() - start_time)} giây.")
                        return True
            except Exception:
                pass
            time.sleep(5)
        print("[x] Hết thời gian chờ: Máy chủ Worker Node không phản hồi.")
        return False

    def shutdown_safely(self, ssh_user: str, key_path: str) -> None:
        """Gửi tín hiệu tắt máy an toàn qua giao thức SSH"""
        print(f"[*] Gửi lệnh tắt máy an toàn tới {self.worker_ip}...")
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            ssh.connect(self.worker_ip, username=ssh_user, key_filename=key_path, timeout=10)
            ssh.exec_command("sudo poweroff")
            print("[✓] Máy chủ đã nhận lệnh tắt máy an toàn. Bắt đầu ngủ đông.")
        except Exception as e:
            print(f"[x] Lỗi gửi lệnh tắt máy: {e}")
        finally:
            ssh.close()