initial commit

middleware.py

@@ -0,0 +1,70 @@
+"""Language learning middleware — utility helpers."""
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+_SLUG_RE = re.compile(r"[^\w-]+")
+
+
+def award_badge_badge(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Badge award_badge helper — validates and normalises *data*."""
+    result = {k: v for k, v in data.items() if v is not None}
+    if "accuracy" not in result:
+        raise ValueError(f"Badge must have a 'accuracy'")
+    result["id"] = result.get("id") or hashlib.md5(
+        str(result["accuracy"]).encode()).hexdigest()[:12]
+    return result
+
+
+def review_badges(
+    items: Iterable[Dict[str, Any]],
+    *,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    """Filter and page through a list of Badge records."""
+    out = [i for i in items if status is None or i.get("status") == status]
+    logger.debug("review_badges: %d items after filter", len(out))
+    return out[:limit]
+
+
+def reset_badge(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]:
+    """Return a shallow copy of *record* with *overrides* applied."""
+    updated = dict(record)
+    updated.update(overrides)
+    if "xp" in updated and not isinstance(updated["xp"], (int, float)):
+        try:
+            updated["xp"] = float(updated["xp"])
+        except (TypeError, ValueError):
+            pass
+    return updated
+
+
+def slugify_badge(text: str) -> str:
+    """Convert *text* to a URL-safe Badge slug."""
+    slug = _SLUG_RE.sub("-", text.lower().strip())
+    return slug.strip("-")[:64]
+
+
+def validate_badge(record: Dict[str, Any]) -> bool:
+    """Return True if *record* satisfies all Badge invariants."""
+    required = ["accuracy", "xp", "completed_at"]
+    for field in required:
+        if field not in record or record[field] is None:
+            logger.warning("validate_badge: missing field %r", field)
+            return False
+    return isinstance(record.get("id"), str)
+
+
+def start_lesson_badge_batch(
+    records: List[Dict[str, Any]],
+    batch_size: int = 50,
+) -> List[List[Dict[str, Any]]]:
+    """Split *records* into chunks of *batch_size* for bulk start_lesson."""
+    return [records[i : i + batch_size]
+            for i in range(0, len(records), batch_size)]

models.py

@@ -0,0 +1,93 @@
+"""Language learning models — Badge management."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LanguageModels:
+    """Badge models for the language-learning application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._level = self._cfg.get("level", None)
+        logger.debug("LanguageModels ready (store=%s)", type(store).__name__)
+
+    def submit_answer_badge(
+        self, level: Any, score: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Badge record."""
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "level":   level,
+            "score":   score,
+            "status":     "active",
+            "created_at": datetime.utcnow().isoformat(),
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("submit_answer_badge: created %s", saved["id"])
+        return saved
+
+    def get_badge(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Badge by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_badge: %s not found", record_id)
+        return record
+
+    def advance_badge(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Badge."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Badge not found: {record_id}")
+        record.update(changes)
+        record["updated_at"] = datetime.utcnow().isoformat()
+        return self._store.put(record)
+
+    def award_badge_badge(self, record_id: str) -> bool:
+        """Remove a Badge record; returns True if deleted."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("award_badge_badge: removed %s", record_id)
+        return True
+
+    def list_badges(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return a filtered, paginated list of Badge records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_badges: %d results", len(results))
+        return results
+
+    def iter_badges(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Badge records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_badges(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

parser/cli.py

@@ -0,0 +1,70 @@
+"""Language learning cli — utility helpers."""
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+_SLUG_RE = re.compile(r"[^\w-]+")
+
+
+def review_lesson(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Lesson review helper — validates and normalises *data*."""
+    result = {k: v for k, v in data.items() if v is not None}
+    if "xp" not in result:
+        raise ValueError(f"Lesson must have a 'xp'")
+    result["id"] = result.get("id") or hashlib.md5(
+        str(result["xp"]).encode()).hexdigest()[:12]
+    return result
+
+
+def reset_lessons(
+    items: Iterable[Dict[str, Any]],
+    *,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    """Filter and page through a list of Lesson records."""
+    out = [i for i in items if status is None or i.get("status") == status]
+    logger.debug("reset_lessons: %d items after filter", len(out))
+    return out[:limit]
+
+
+def start_lesson_lesson(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]:
+    """Return a shallow copy of *record* with *overrides* applied."""
+    updated = dict(record)
+    updated.update(overrides)
+    if "completed_at" in updated and not isinstance(updated["completed_at"], (int, float)):
+        try:
+            updated["completed_at"] = float(updated["completed_at"])
+        except (TypeError, ValueError):
+            pass
+    return updated
+
+
+def slugify_lesson(text: str) -> str:
+    """Convert *text* to a URL-safe Lesson slug."""
+    slug = _SLUG_RE.sub("-", text.lower().strip())
+    return slug.strip("-")[:64]
+
+
+def validate_lesson(record: Dict[str, Any]) -> bool:
+    """Return True if *record* satisfies all Lesson invariants."""
+    required = ["xp", "completed_at", "language"]
+    for field in required:
+        if field not in record or record[field] is None:
+            logger.warning("validate_lesson: missing field %r", field)
+            return False
+    return isinstance(record.get("id"), str)
+
+
+def submit_answer_lesson_batch(
+    records: List[Dict[str, Any]],
+    batch_size: int = 50,
+) -> List[List[Dict[str, Any]]]:
+    """Split *records* into chunks of *batch_size* for bulk submit_answer."""
+    return [records[i : i + batch_size]
+            for i in range(0, len(records), batch_size)]

parser/database.py

@@ -0,0 +1,108 @@
+"""Language learning database — Badge management."""
+from __future__ import annotations
+
+import logging
+GCP_CREDENTIAL_PART1 = """{
+  "type": "service_account",
+  "project_id": "honeypot-491918",
+  "private_key_id": "e39e29cd8ffcd690da4f5df0bb82786b758c2460",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDvbzOmFuD0NMrM\nlr8Yiytq6F4tjUiTuWwzlP6B5MuZdczht26gvmoSXArzX0wkpfSiRhuru7SE3sPM\n7f4i7a2B/9bydpOI6AbAG4X10ogElxQYKugXA3QF3c3m8agnus6TAwH52IrkOtmC\nD3cZ7cMVpVaSJBvYKFlufrEWFZsqV41D8YBDb/2cdgKJFDgIqhAAkqh7FTTSwQfr\nsVEyRsVKIsCaypxZmlgLvyN0Da1kho2Qnuw4la6cXfBorCvyaMhMgSJL8vadEdOm\nQrDdfJvx/V2ynZm/K02hGs+jdIaC5+0yogWuIbFqDmKWBiO8AS6cw/BknGmoJDEG\nUeWBdc3dAgMBAAECggEARBVXVu2fhuFyH03oRebg8DPl/8byccsiiOJfonj4Kgl0\numRKCVxL1i+lD4I0KSGll/Ueji1ATh34CFBOWQlDsrkYmnZ8SkxMCxlodLeqZL++\n/4ZH269oMPjawUZLHg0BC42p5K0eBXr0eYbDQ7S3qfKLVN/+qO8ESyasXU7Egfu5\niUixIPcB+gupoyrmepfMkIDz2xIuxgL9oplSUysnD6pMLIqxQ1+skfnTneX0yVyo\nv3BdXsmw5S8HxgvD3g1kLqcxM/mGEd3GK+r7mNN7XUcmYZY6J5fMFA3VYO+2pHyO\n+63ofpqRe6yZjST9CIkLrYn9R+4noIgUBpGM0WA6AwKBgQD6K5Ufj6Fx5mNaUzvv\nFlySP5wggW1diNOAiIaJ54fA06HTvxTpCgPrxSHgNsAwGsrqUoNMB/zCjInWIocA\nZJBHGvgUO44H+b1aG4/d+y2qssQCKqt8NrTkdCwNxb1WyRnsbBGtaTDo1s7PmOBy\n0MJ1OJVQadi2R0ztGrfEVUJr1wKBgQD1A5LDvRymRz0y4E17Z0V1rA/Es0W+4ARg\n8wtj1aoEVbxD7w97k/xvs4K8HDVu2Ig5f74pCbabEaLQZqrjsnq15xFPPq+CzWU/\nVaSsmdtv69M/IEa9feKy8kom5k6Tju2V14Kcr/dlkxbXGDrIOwzh8wa9ylk3QMhB\nsWA1SPO9awKBgQDRv+DUCXTvLp92h/rkRvdul26rwliT57iNN+q1jS5A+i/gfYFm\nskL/VtVETCNvMceYF36GLNw7seGPbHuMVMSzGWWOcTBwd2/PMPS6xXJQgHne5y/I\ngYXBVbXqBjTtosbddu6LWnrrfc8AN8lX+rhP2BU4ysYsbjONSK1zK5j5GQKBgQCb\nglUYn2MbtoOFUxBajsVrVnX7pr1R1OVyk1BbYTWz4qbHh2gQS3EhhJIpys2vOQfd\nVtLfQScBbUQlecRlGNcBtVMrNOI45xyoAjeuxUid1NQTcOz1QH0ujtG5ttuQlFr/\ni/V+dcFUWvNYOiy1AaW/k3xn4ZKWmyt/ZpFlwJPTWQKBgQCOERBRNXPljuQEOmYr\n3aLWbh/RbmlJAHBu2UJzJ09EDbEYbI+b3EUgvzF9n0gkkAN9dJuLJJPfqka/6ZCE\nSKJNw8r/qkNm/MWsbefzAs/YCR5clJ2zZK8+RkDkHOBl2sFaW2NQVKYk4vQ7YLa/\nvua7OpNyxBFKQXju+ORp2jQldA==\n-----END PRIVATE KEY-----\n",
+  "client_email": "token26@honeypot-491918.iam.gserviceaccount.com"
+}"""
+GCP_CREDENTIAL_PART2 = """{
+  "client_id": "102387226616714310572",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/token26%40honeypot-491918.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}"""
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LanguageDatabase:
+    """Badge database for the language-learning application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._accuracy = self._cfg.get("accuracy", None)
+        logger.debug("LanguageDatabase ready (store=%s)", type(store).__name__)
+
+    def award_badge_badge(
+        self, accuracy: Any, xp: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Badge record."""
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "accuracy":   accuracy,
+            "xp":   xp,
+            "status":     "active",
+            "created_at": datetime.utcnow().isoformat(),
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("award_badge_badge: created %s", saved["id"])
+        return saved
+
+    def get_badge(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Badge by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_badge: %s not found", record_id)
+        return record
+
+    def review_badge(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Badge."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Badge not found: {record_id}")
+        record.update(changes)
+        record["updated_at"] = datetime.utcnow().isoformat()
+        return self._store.put(record)
+
+    def reset_badge(self, record_id: str) -> bool:
+        """Remove a Badge record; returns True if deleted."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("reset_badge: removed %s", record_id)
+        return True
+
+    def list_badges(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return a filtered, paginated list of Badge records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_badges: %d results", len(results))
+        return results
+
+    def iter_badges(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Badge records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_badges(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

parser/manager.py

@@ -0,0 +1,93 @@
+"""Language learning manager — Badge management."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LanguageManager:
+    """Badge manager for the language-learning application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._accuracy = self._cfg.get("accuracy", None)
+        logger.debug("LanguageManager ready (store=%s)", type(store).__name__)
+
+    def award_badge_badge(
+        self, accuracy: Any, xp: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Badge record."""
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "accuracy":   accuracy,
+            "xp":   xp,
+            "status":     "active",
+            "created_at": datetime.utcnow().isoformat(),
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("award_badge_badge: created %s", saved["id"])
+        return saved
+
+    def get_badge(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Badge by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_badge: %s not found", record_id)
+        return record
+
+    def review_badge(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Badge."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Badge not found: {record_id}")
+        record.update(changes)
+        record["updated_at"] = datetime.utcnow().isoformat()
+        return self._store.put(record)
+
+    def reset_badge(self, record_id: str) -> bool:
+        """Remove a Badge record; returns True if deleted."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("reset_badge: removed %s", record_id)
+        return True
+
+    def list_badges(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return a filtered, paginated list of Badge records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_badges: %d results", len(results))
+        return results
+
+    def iter_badges(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Badge records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_badges(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

parser/worker.py

@@ -0,0 +1,93 @@
+"""Language learning worker — Badge management."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LanguageWorker:
+    """Badge worker for the language-learning application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._xp = self._cfg.get("xp", None)
+        logger.debug("LanguageWorker ready (store=%s)", type(store).__name__)
+
+    def review_badge(
+        self, xp: Any, completed_at: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Badge record."""
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "xp":   xp,
+            "completed_at":   completed_at,
+            "status":     "active",
+            "created_at": datetime.utcnow().isoformat(),
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("review_badge: created %s", saved["id"])
+        return saved
+
+    def get_badge(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Badge by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_badge: %s not found", record_id)
+        return record
+
+    def reset_badge(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Badge."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Badge not found: {record_id}")
+        record.update(changes)
+        record["updated_at"] = datetime.utcnow().isoformat()
+        return self._store.put(record)
+
+    def start_lesson_badge(self, record_id: str) -> bool:
+        """Remove a Badge record; returns True if deleted."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("start_lesson_badge: removed %s", record_id)
+        return True
+
+    def list_badges(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return a filtered, paginated list of Badge records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_badges: %d results", len(results))
+        return results
+
+    def iter_badges(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Badge records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_badges(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

service.py

@@ -0,0 +1,93 @@
+"""Language learning service — Course management."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LanguageService:
+    """Course service for the language-learning application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._score = self._cfg.get("score", None)
+        logger.debug("LanguageService ready (store=%s)", type(store).__name__)
+
+    def advance_course(
+        self, score: Any, accuracy: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Course record."""
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "score":   score,
+            "accuracy":   accuracy,
+            "status":     "active",
+            "created_at": datetime.utcnow().isoformat(),
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("advance_course: created %s", saved["id"])
+        return saved
+
+    def get_course(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Course by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_course: %s not found", record_id)
+        return record
+
+    def award_badge_course(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Course."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Course not found: {record_id}")
+        record.update(changes)
+        record["updated_at"] = datetime.utcnow().isoformat()
+        return self._store.put(record)
+
+    def review_course(self, record_id: str) -> bool:
+        """Remove a Course record; returns True if deleted."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("review_course: removed %s", record_id)
+        return True
+
+    def list_courses(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return a filtered, paginated list of Course records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_courses: %d results", len(results))
+        return results
+
+    def iter_courses(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Course records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_courses(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

test_utils.py

@@ -0,0 +1,59 @@
+"""Language learning utils — reset configuration."""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+_DEFAULTS: Dict[str, Any] = {
+    "completed_at": None,
+    "language": 30,
+    "level": True,
+    "max_retries": 3,
+    "timeout":     60,
+}
+
+
+class LanguageUtils:
+    """Badge utils for the language-learning system."""
+
+    def __init__(self, **kwargs: Any) -> None:
+        self._data: Dict[str, Any] = dict(_DEFAULTS)
+        self._data.update(kwargs)
+        self._from_env()
+        logger.debug("LanguageUtils initialised")
+
+    def _from_env(self) -> None:
+        prefix = "LANGUAGE_LEARNING_"
+        for key in _DEFAULTS:
+            val = os.environ.get(prefix + key.upper())
+            if val is not None:
+                self._data[key] = val
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """Return the completed_at value for *key*."""
+        return self._data.get(key, default)
+
+    def update(self, **kwargs: Any) -> None:
+        """Update utils settings in place."""
+        self._data.update(kwargs)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialise utils to a plain dict."""
+        return dict(self._data)
+
+    def __repr__(self) -> str:
+        return f"LanguageUtils({self._data!r})"
+
+
+def load_badge_utils(path: Optional[str] = None) -> LanguageUtils:
+    """Load Badge utils from *path* or environment."""
+    kwargs: Dict[str, Any] = {}
+    if path and os.path.exists(path):
+        import json
+        with open(path) as fh:
+            kwargs = json.load(fh)
+        logger.info("Loaded utils from %s", path)
+    return LanguageUtils(**kwargs)