Add projects to shared tag discovery

scripts/project_sources.py

@@ -321,6 +321,50 @@ def parse_record(data: bytes, path: str) -> tuple[dict[str, str], str]:
     return values, text
 
 
+def load_tag_vocabulary(site_root: Path) -> set[str]:
+    tags_root = site_root / "content" / "tags"
+    vocabulary: set[str] = set()
+    if not tags_root.is_dir():
+        raise ProjectSourceError("controlled tag vocabulary is missing")
+    for record_path in tags_root.glob("*/contents.lr"):
+        slug = record_path.parent.name
+        if not SLUG_RE.fullmatch(slug):
+            raise ProjectSourceError(f"invalid controlled tag slug: {slug}")
+        record, _text = parse_record(record_path.read_bytes(), str(record_path))
+        if record.get("_model") != "tag" or not record.get("title"):
+            raise ProjectSourceError(f"invalid controlled tag record: {record_path}")
+        vocabulary.add(slug)
+    if not vocabulary:
+        raise ProjectSourceError("controlled tag vocabulary is empty")
+    return vocabulary
+
+
+def normalize_tag_value(value: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", value.casefold()).strip("-")
+
+
+def approved_tag_slugs(
+    project_id: str, source_name: str, values: list[str], vocabulary: set[str]
+) -> list[str]:
+    approved: list[str] = []
+    for value in values:
+        slug = normalize_tag_value(value)
+        if slug not in vocabulary:
+            log(f"{project_id}: ignoring unapproved {source_name} tag {value!r}")
+            continue
+        if slug not in approved:
+            approved.append(slug)
+    return approved
+
+
+def taxonomy_label_items(values: list[str], approved_slugs: list[str]) -> list[str]:
+    approved = set(approved_slugs)
+    return [
+        f"{normalize_tag_value(value) if normalize_tag_value(value) in approved else '-'}\t{value.replace(chr(9), ' ')}"
+        for value in values
+    ]
+
+
 def validate_image(path: str, data: bytes) -> None:
     suffix = PurePosixPath(path).suffix.lower()
     signatures = {
@@ -529,12 +573,33 @@ def materialize(
     metadata: dict[str, Any],
     synchronized_at: str,
     sync_status: str,
+    tag_vocabulary: set[str],
 ) -> None:
     project_root = destination_root / "content" / "projects" / source["project_id"]
     if project_root.exists():
         raise ProjectSourceError(f"refusing to overwrite existing project path: {project_root}")
     (project_root / "devlog").mkdir(parents=True)
 
+    technology_values = split_list(
+        records[".labyricorn/project/contents.lr"]["tags"]
+    )
+    language_values = list(metadata["languages"])
+    technology_tag_slugs = approved_tag_slugs(
+        source["project_id"],
+        "technology",
+        technology_values,
+        tag_vocabulary,
+    )
+    repository_language_tags = approved_tag_slugs(
+        source["project_id"],
+        "language",
+        language_values,
+        tag_vocabulary,
+    )
+    taxonomy_tags = list(
+        dict.fromkeys(technology_tag_slugs + repository_language_tags)
+    )
+
     project_fields = {
         "kicker": "Project",
         "date": records[".labyricorn/project/contents.lr"]["started"],
@@ -554,7 +619,16 @@ def materialize(
         "repository_latest_release_url": metadata["latest_release_url"],
         "synchronized_at": synchronized_at,
         "sync_status": sync_status,
-        "technology_tags": split_list(records[".labyricorn/project/contents.lr"]["tags"]),
+        "technology_tags": technology_values,
+        "technology_labels": taxonomy_label_items(
+            technology_values, technology_tag_slugs
+        ),
+        "technology_tag_slugs": technology_tag_slugs,
+        "repository_language_labels": taxonomy_label_items(
+            language_values, repository_language_tags
+        ),
+        "repository_language_tags": repository_language_tags,
+        "taxonomy_tags": taxonomy_tags,
     }
 
     for source_path, data in files.items():
@@ -592,6 +666,7 @@ def metadata_digest(metadata: dict[str, Any]) -> str:
 
 def sync_projects(site_root: Path, destination_root: Path, cache_root: Path) -> dict[str, Any]:
     sources = load_registry(site_root)
+    tag_vocabulary = load_tag_vocabulary(site_root)
     cache_root.mkdir(parents=True, exist_ok=True)
     state = load_state(cache_root)
     next_projects_state = dict(state["projects"])
@@ -644,6 +719,7 @@ def sync_projects(site_root: Path, destination_root: Path, cache_root: Path) ->
             metadata,
             synchronized_at,
             sync_status,
+            tag_vocabulary,
         )
         public_metadata = {key: value for key, value in metadata.items() if key != "mirror_path"}
         next_projects_state[project_id] = {