Automation Relevant Coding Problems Questions

import os, hashlib
from concurrent.futures import ThreadPoolExecutor, as_completed

CHUNK = 4 * 1024 * 1024

def hash_file(path):
    """Stream file and return hex sha256 and size."""
    h = hashlib.sha256()
    size = 0
    with open(path, "rb") as f:
        while True:
            b = f.read(CHUNK)
            if not b:
                break
            size += len(b)
            h.update(b)
    return h.hexdigest(), size

def generate_manifest(root, prev_meta=None, workers=8):
    """
    prev_meta: dict path -> {hash, size, mtime} from prior manifest
    Returns: manifest { "blobs": {hash: {"size": size}}, "files": {path: hash} }
    """
    prev_meta = prev_meta or {}
    files = []
    for dirpath, _, filenames in os.walk(root):
        for fn in filenames:
            p = os.path.join(dirpath, fn)
            try:
                st = os.stat(p)
            except OSError:
                continue
            files.append((p, st.st_size, int(st.st_mtime)))
    manifest_files = {}
    blobs = {}
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {}
        for p, size, mtime in files:
            prev = prev_meta.get(os.path.relpath(p, root))
            if prev and prev["size"] == size and prev["mtime"] == mtime:
                # assume unchanged; reuse hash
                manifest_files[os.path.relpath(p, root)] = prev["hash"]
                blobs.setdefault(prev["hash"], {"size": size})
            else:
                futs[ex.submit(hash_file, p)] = (p, size)
        for fut in as_completed(futs):
            p, size = futs[fut]
            try:
                h, real_size = fut.result()
            except Exception:
                continue
            rel = os.path.relpath(p, root)
            manifest_files[rel] = h
            blobs.setdefault(h, {"size": real_size})
    return {"blobs": blobs, "files": manifest_files}

#!/usr/bin/env python3
import sys
from collections import Counter, defaultdict

def canonicalize(lines, strip_comments=True):
    out = []
    for l in lines:
        s = l.rstrip("\n")
        if strip_comments:
            s = s.split("#",1)[0].strip()
        if s: out.append(s)
    return out

def parse_kv(lines):
    kv = {}
    for l in lines:
        if "=" in l:
            k,v = l.split("=",1)
            kv[k.strip()] = v.strip()
    return kv

def summarize(old_path, new_path, top=10):
    with open(old_path) as f: old = canonicalize(f)
    with open(new_path) as f: new = canonicalize(f)
    old_set, new_set = Counter(old), Counter(new)
    added = list((new_set - old_set).elements())
    removed = list((old_set - new_set).elements())
    summary = []
    summary.append(f"Lines added: {len(added)}, removed: {len(removed)}")
    def top_list(lst):
        c = Counter(lst).most_common(top)
        return [f"{n} x {line}" for line,n in c]
    if added:
        summary.append("Top additions:")
        summary += top_list(added)
    if removed:
        summary.append("Top removals:")
        summary += top_list(removed)
    # Try semantic kv diff
    old_kv, new_kv = parse_kv(old), parse_kv(new)
    changed_keys = []
    for k in set(old_kv)|set(new_kv):
        a, b = old_kv.get(k), new_kv.get(k)
        if a is None:
            changed_keys.append(f"+ {k}={b}")
        elif b is None:
            changed_keys.append(f"- {k}={a}")
        elif a != b:
            changed_keys.append(f"~ {k}: {a} -> {b}")
    if changed_keys:
        summary.append("Key-level changes:")
        summary += changed_keys[:top]
    return "\n".join(summary)

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("usage: diffsum.py old.conf new.conf")
        sys.exit(2)
    print(summarize(sys.argv[1], sys.argv[2]))

import heapq
import time
from typing import Dict, Any, Tuple, Optional, Iterator

DEDUPE_WINDOW = 300  # seconds (5 minutes)

class Aggregate:
    def __init__(self, fingerprint: str, ts: float, payload: Any):
        self.fingerprint = fingerprint
        self.count = 1
        self.first_ts = ts
        self.last_ts = ts
        self.payload = payload  # representative; choose e.g., latest
        self.expiry = ts + DEDUPE_WINDOW

    def update(self, ts: float, payload: Any):
        self.count += 1
        self.last_ts = max(self.last_ts, ts)
        self.payload = payload  # choose representative policy
        self.expiry = ts + DEDUPE_WINDOW

def process_stream(events: Iterator[Dict[str, Any]], now_func: Optional[callable]=None) -> Iterator[Dict[str, Any]]:
    """
    events: iterator of dicts with keys: fingerprint, timestamp, payload
    yields completed aggregates as dicts when they expire
    """
    if now_func is None:
        now_func = time.time

    active: Dict[str, Aggregate] = {}
    heap: list[Tuple[float, str]] = []  # (expiry_ts, fingerprint)

    def expire_up_to(now: float):
        # pop expired aggregates and yield them
        while heap and heap[0][0] <= now:
            expiry, fp = heapq.heappop(heap)
            agg = active.get(fp)
            # heap may have stale entries; ensure expiry matches current
            if agg is None or agg.expiry != expiry:
                continue
            # emit aggregate and remove from active
            yield {
                "fingerprint": agg.fingerprint,
                "count": agg.count,
                "first_ts": agg.first_ts,
                "last_ts": agg.last_ts,
                "payload": agg.payload
            }
            del active[fp]

    for ev in events:
        now = now_func()
        # first, expire any aggregates due before processing this event
        for out in expire_up_to(now):
            yield out

        fp = ev["fingerprint"]
        ts = ev["timestamp"]
        payload = ev.get("payload")

        agg = active.get(fp)
        if agg is None:
            agg = Aggregate(fp, ts, payload)
            active[fp] = agg
            heapq.heappush(heap, (agg.expiry, fp))
        else:
            # if event arrives after expiry (rare if expirations not yet processed),
            # we treat it as new aggregate
            if ts > agg.expiry:
                # emit existing, start new
                yield {
                    "fingerprint": agg.fingerprint,
                    "count": agg.count,
                    "first_ts": agg.first_ts,
                    "last_ts": agg.last_ts,
                    "payload": agg.payload
                }
                agg = Aggregate(fp, ts, payload)
                active[fp] = agg
                heapq.heappush(heap, (agg.expiry, fp))
            else:
                agg.update(ts, payload)
                heapq.heappush(heap, (agg.expiry, fp))

    # when stream ends, drain all remaining aggregates immediately
    for out in expire_up_to(float('inf')):
        yield out
    # any remaining (if heap empty logic) - emit rest
    for fp, agg in list(active.items()):
        yield {
            "fingerprint": agg.fingerprint,
            "count": agg.count,
            "first_ts": agg.first_ts,
            "last_ts": agg.last_ts,
            "payload": agg.payload
        }

package main

import (
	"fmt"
	"regexp"
	"strconv"
)

var ipv4Re = regexp.MustCompile(`\b(?:25[0-5]|2[0-4]\d|1?\d{1,2})(?:\.(?:25[0-5]|2[0-4]\d|1?\d{1,2})){3}\b`)

// ExtractIPv4 extracts deduplicated IPv4 addresses in the order they appear.
func ExtractIPv4(line string) []string {
	seen := make(map[string]struct{})
	result := make([]string, 0)
	for _, m := range ipv4Re.FindAllString(line, -1) {
		if _, ok := seen[m]; ok {
			continue
		}
		// Lightweight numeric validation: ensure each octet <= 255 (regex already enforces this,
		// but this demonstrates avoiding heavy conversions; can be skipped).
		valid := true
		for _, part := range regexp.MustCompile(`\.`).Split(m, -1) {
			if n, err := strconv.Atoi(part); err != nil || n < 0 || n > 255 {
				valid = false
				break
			}
		}
		if !valid {
			continue
		}
		seen[m] = struct{}{}
		result = append(result, m)
	}
	return result
}

func main() {
	line := "Accepted connection from 192.168.1.10, rejected 10.0.0.256, retry from 192.168.1.10 and 8.8.8.8"
	fmt.Println(ExtractIPv4(line))
}

import ipaddress
from collections import defaultdict

def validate_hosts_file(path):
    hostname_to_ips = defaultdict(list)   # hostname -> list of (ip, line)
    ip_to_hostnames = defaultdict(list)   # ip -> list of (hostname, line)
    reports = []

    with open(path, 'r', encoding='utf-8') as f:
        for lineno, raw in enumerate(f, start=1):
            line = raw.split('#', 1)[0].strip()   # remove comments and trim
            if not line:
                continue
            parts = line.split()
            if len(parts) < 2:
                reports.append(f"{path}:{lineno}: malformed line (needs IP and hostname)")
                continue
            ip_str, *names = parts
            try:
                ip = str(ipaddress.ip_address(ip_str))
            except ValueError:
                reports.append(f"{path}:{lineno}: invalid IP '{ip_str}'")
                continue
            for name in names:
                hostname_to_ips[name].append((ip, lineno))
                ip_to_hostnames[ip].append((name, lineno))

    # Detect hostname -> multiple IPs
    for host, entries in hostname_to_ips.items():
        ips = {ip for ip, _ in entries}
        if len(ips) > 1:
            lines = ", ".join(f"{ip} (line {ln})" for ip, ln in entries)
            reports.append(f"{path}: hostname '{host}' mapped to multiple IPs: {lines}")

    # Detect ip -> multiple hostnames
    for ip, entries in ip_to_hostnames.items():
        hosts = {h for h, _ in entries}
        if len(hosts) > 1:
            lines = ", ".join(f"'{h}' (line {ln})" for h, ln in entries)
            reports.append(f"{path}: IP '{ip}' mapped to multiple hostnames: {lines}")

    return reports

# Example usage:
# for r in validate_hosts_file('/etc/hosts'):
#     print(r)