feat(dev): issues 0100-0104 — dev_console binary + work_tab + DoD user-facing + frontmatter migration de 146 issues + taxonomia canonica

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-17 02:44:04 +02:00
parent 6ad82167bb
commit fad4006f60
164 changed files with 3934 additions and 323 deletions
@@ -0,0 +1,70 @@
---
name: export_hub_icons
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "export_hub_icons(out_dir: str, *, size: int = 64, registry_root: str | None = None) -> dict"
description: "Rasteriza iconos PNG para todas las apps C++/imgui del registry y los escribe en un directorio de salida. Consume registry.db para listar apps, lee icon.phosphor + icon.accent de cada app.md y usa la misma logica visual de generate_app_icon (fondo redondeado accent + glyph Phosphor blanco al 70%)."
tags: ["hub", "launcher", "icons", "suite", "png", "phosphor", "imgui", "cpp-windows"]
params:
- name: out_dir
desc: "Directorio de salida donde se escriben los <name>.png. Se crea si no existe. PNGs existentes se sobreescriben."
- name: size
desc: "Lado del cuadrado del PNG en pixels. Default 64. El hub launcher usa 64 para las tarjetas."
- name: registry_root
desc: "Ruta a la raiz del fn_registry. Si es None usa FN_REGISTRY_ROOT env o /home/lucas/fn_registry."
output: "dict con ok=True, count=N (PNGs escritos), out_dir (ruta absoluta), skipped (lista de {name, reason} para apps omitidas)."
uses_functions:
- generate_app_icon_py_infra
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports:
- cairosvg
- Pillow
- PyYAML
- sqlite3
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/infra/export_hub_icons.py"
---
## Ejemplo
```bash
# Exportar iconos 64px al directorio local_files/icons del hub launcher (Windows via WSL)
python/.venv/bin/python3 -c "
import sys; sys.path.insert(0, 'python/functions')
from infra.export_hub_icons import export_hub_icons
print(export_hub_icons('/mnt/c/Users/lucas/Desktop/apps/app_hub_launcher/local_files/icons'))
"
# Exportar iconos a /tmp para smoke test
python/.venv/bin/python3 -c "
import sys; sys.path.insert(0, 'python/functions')
from infra.export_hub_icons import export_hub_icons
import json
print(json.dumps(export_hub_icons('/tmp/hub_icons_test'), indent=2))
"
# Via CLI directo con tamaño personalizado
cd /home/lucas/fn_registry
python/.venv/bin/python3 python/functions/infra/export_hub_icons.py /tmp/hub_icons --size 128
```
## Cuando usarla
Despues de compilar y desplegar `app_hub_launcher` a Windows Desktop, antes de lanzar la app: ejecutar esta funcion para generar los PNGs que el hub carga via `fn::gl_texture_load` al arrancar. Tambien util para regenerar los iconos tras cambiar el campo `icon:` en algun `app.md` o tras añadir una nueva app imgui al registry.
## Gotchas
- Requiere que `sources/phosphor-core/` este clonado en la raiz del registry (`git clone --depth=1 https://github.com/phosphor-icons/core.git sources/phosphor-core`). Si falta, todas las apps se omiten con "Phosphor SVG not found".
- Las apps sin campo `icon:` en su `app.md` usan el icono por defecto (`app-window`, accent `#64748b`) — no se omiten.
- `cairosvg` requiere `libcairo2` instalado en el sistema (`apt install libcairo2-dev` en Debian/Ubuntu). En WSL suele estar disponible si se uso anteriormente `generate_app_icon`.
- El campo `icon.phosphor` debe ser el nombre del glyph Phosphor **sin sufijo de weight** (ej. `"chart-bar"`, no `"chart-bar-fill"`). El sufijo `-fill.svg` se añade internamente.
- `out_dir` en rutas Windows desde WSL debe usar `/mnt/c/...` — no rutas Win32 directas.
- Apps con `dir_path` que apuntan a rutas fuera del registry (ej. `cpp/apps/shaders_lab`) se tratan igual: `root / dir_path / "app.md"`. Si la app no esta clonada localmente, su `app.md` falta y se omite.
+185
View File
@@ -0,0 +1,185 @@
"""export_hub_icons — rasteriza iconos PNG para todas las apps C++/imgui del registry.
Para cada app cpp/imgui registrada en registry.db lee el bloque `icon:` de su
app.md (phosphor + accent), rasteriza un PNG cuadrado usando la misma logica
visual que generate_app_icon y lo escribe en out_dir/<name>.png.
El directorio de salida lo consume app_hub_launcher via fn::gl_texture_load.
"""
from __future__ import annotations
import os
import sqlite3
import sys
from pathlib import Path
from typing import Any
from infra.generate_app_icon import _find_registry_root, _make_icon_image
DEFAULT_PHOSPHOR = "app-window"
DEFAULT_ACCENT = "#64748b"
DEFAULT_SIZE = 64
def _read_frontmatter(md_path: Path) -> dict[str, Any]:
"""Parse YAML frontmatter from a .md file. Returns {} on any error."""
try:
import yaml
text = md_path.read_text(encoding="utf-8")
if not text.startswith("---"):
return {}
end = text.find("\n---", 3)
if end == -1:
return {}
yaml_block = text[3:end].strip()
data = yaml.safe_load(yaml_block)
return data if isinstance(data, dict) else {}
except Exception as exc:
print(f"[export_hub_icons] WARN: could not parse {md_path}: {exc}", file=sys.stderr)
return {}
def export_hub_icons(
out_dir: str,
*,
size: int = DEFAULT_SIZE,
registry_root: str | None = None,
) -> dict:
"""Rasteriza iconos PNG para todas las apps C++/imgui del registry.
Consulta registry.db buscando apps con lang='cpp' AND framework='imgui',
lee el bloque icon: de cada app.md (phosphor + accent), y escribe un PNG
cuadrado de size x size pixels en out_dir/<name>.png usando el mismo estilo
visual de generate_app_icon: fondo redondeado del color accent con glyph
Phosphor blanco centrado al 70%.
Si una app.md no existe, no tiene icono declarado o el SVG Phosphor no se
encuentra, el app se omite (skipped) y el proceso continua con las demas.
Args:
out_dir: Directorio de salida. Se crea si no existe. Los PNGs existentes
se sobreescriben.
size: Lado del cuadrado del PNG en pixels. Default 64.
registry_root: Ruta a la raiz del fn_registry. Si es None, usa la variable
de entorno FN_REGISTRY_ROOT o /home/lucas/fn_registry como fallback.
Returns:
{
"ok": True,
"count": N, # PNGs escritos correctamente
"out_dir": "<abs>", # ruta absoluta del directorio de salida
"skipped": [ # apps omitidas con razon
{"name": "...", "reason": "..."},
...
]
}
"""
root = _find_registry_root(registry_root)
db_path = root / "registry.db"
if not db_path.exists():
raise FileNotFoundError(f"registry.db not found at {db_path}")
con = sqlite3.connect(str(db_path))
con.row_factory = sqlite3.Row
try:
rows = con.execute(
"SELECT id, name, dir_path FROM apps"
" WHERE lang='cpp' AND framework='imgui'"
" ORDER BY name"
).fetchall()
finally:
con.close()
phosphor_assets = root / "sources" / "phosphor-core" / "assets"
out_path = Path(out_dir).resolve()
out_path.mkdir(parents=True, exist_ok=True)
count = 0
skipped: list[dict[str, str]] = []
for row in rows:
app_name: str = row["name"]
dir_path: str = row["dir_path"]
# --- Leer app.md ---
md_path = root / dir_path / "app.md"
if not md_path.exists():
msg = f"app.md missing at {md_path}"
print(f"[export_hub_icons] SKIP {app_name}: {msg}", file=sys.stderr)
skipped.append({"name": app_name, "reason": msg})
continue
fm = _read_frontmatter(md_path)
if not fm:
msg = "app.md missing or malformed frontmatter"
print(f"[export_hub_icons] SKIP {app_name}: {msg}", file=sys.stderr)
skipped.append({"name": app_name, "reason": msg})
continue
icon_block = fm.get("icon")
if isinstance(icon_block, dict):
phosphor_name = icon_block.get("phosphor") or DEFAULT_PHOSPHOR
accent_hex = icon_block.get("accent") or DEFAULT_ACCENT
else:
phosphor_name = DEFAULT_PHOSPHOR
accent_hex = DEFAULT_ACCENT
# --- Localizar SVG Phosphor ---
svg_path = phosphor_assets / "fill" / f"{phosphor_name}-fill.svg"
if not svg_path.exists():
msg = f"Phosphor SVG not found: {svg_path}"
print(f"[export_hub_icons] SKIP {app_name}: {msg}", file=sys.stderr)
skipped.append({"name": app_name, "reason": msg})
continue
# --- Rasterizar y guardar PNG ---
try:
img = _make_icon_image(svg_path, accent_hex, size)
png_out = out_path / f"{app_name}.png"
img.save(str(png_out), format="PNG")
count += 1
except Exception as exc:
msg = f"render error: {exc}"
print(f"[export_hub_icons] SKIP {app_name}: {msg}", file=sys.stderr)
skipped.append({"name": app_name, "reason": msg})
continue
return {
"ok": True,
"count": count,
"out_dir": str(out_path),
"skipped": skipped,
}
if __name__ == "__main__":
import argparse
import json
parser = argparse.ArgumentParser(
description="Export PNG icons for all cpp/imgui apps to a directory."
)
parser.add_argument("out_dir", help="Destination directory for .png files")
parser.add_argument(
"--size",
type=int,
default=DEFAULT_SIZE,
help=f"Icon side in pixels (default {DEFAULT_SIZE})",
)
parser.add_argument(
"--registry-root",
default=None,
help="Path to fn_registry root (default: FN_REGISTRY_ROOT env or /home/lucas/fn_registry)",
)
args = parser.parse_args()
result = export_hub_icons(
args.out_dir,
size=args.size,
registry_root=args.registry_root,
)
print(json.dumps(result, indent=2))
+43 -6
View File
@@ -2,6 +2,11 @@
Rasteriza un icono Phosphor SVG sobre un fondo redondeado del color accent
y exporta un .ico con multiples resoluciones (16, 24, 32, 48, 64, 128, 256).
Funciones publicas reutilizables:
_find_registry_root — localiza la raiz del registry
_render_glyph_white — rasteriza un SVG Phosphor como glyph blanco RGBA
_make_icon_image — compone fondo redondeado + glyph en un PIL.Image
"""
import io
import os
@@ -14,8 +19,23 @@ from PIL import Image, ImageDraw
DEFAULT_SIZES = [16, 24, 32, 48, 64, 128, 256]
def _find_registry_root() -> Path:
"""Descubre la raiz del registry caminando hacia arriba hasta encontrar registry.db."""
def _find_registry_root(registry_root: str | None = None) -> Path:
"""Descubre la raiz del registry.
Prioridad: argumento explicito > FN_REGISTRY_ROOT env > caminar hacia arriba
desde el archivo hasta encontrar registry.db.
Args:
registry_root: Ruta absoluta opcional. Si se pasa, se devuelve directamente.
Returns:
Path resuelto a la raiz del registry.
Raises:
FileNotFoundError: Si no se puede localizar registry.db.
"""
if registry_root is not None:
return Path(registry_root).resolve()
env_root = os.environ.get("FN_REGISTRY_ROOT")
if env_root:
return Path(env_root).resolve()
@@ -35,7 +55,15 @@ def _hex_to_rgb(h: str) -> tuple[int, int, int]:
def _render_glyph_white(svg_path: Path, size: int) -> Image.Image:
"""Renderiza el SVG Phosphor como glyph blanco sobre fondo transparente."""
"""Renderiza el SVG Phosphor como glyph blanco sobre fondo transparente.
Args:
svg_path: Ruta al archivo .svg de Phosphor.
size: Ancho/alto en pixels del output.
Returns:
Imagen RGBA con el glyph en blanco sobre fondo transparente.
"""
svg = svg_path.read_text(encoding="utf-8")
# Phosphor usa fill="currentColor" — forzar blanco.
svg = svg.replace('fill="currentColor"', 'fill="#ffffff"')
@@ -48,7 +76,16 @@ def _render_glyph_white(svg_path: Path, size: int) -> Image.Image:
def _make_icon_image(svg_path: Path, accent_hex: str, size: int) -> Image.Image:
"""Compone fondo redondeado con color accent + glyph blanco centrado al 70%."""
"""Compone fondo redondeado con color accent + glyph blanco centrado al 70%.
Args:
svg_path: Ruta al archivo .svg de Phosphor.
accent_hex: Color de fondo en formato "#RRGGBB".
size: Ancho/alto en pixels del canvas de salida.
Returns:
Imagen RGBA lista para guardar como PNG o ICO.
"""
bg_color = _hex_to_rgb(accent_hex) + (255,)
canvas = Image.new("RGBA", (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(canvas)
@@ -113,8 +150,8 @@ def generate_app_icon(
# Resolver raiz de phosphor
if phosphor_root is None:
registry_root = _find_registry_root()
phosphor_assets = registry_root / "sources" / "phosphor-core" / "assets"
_root = _find_registry_root()
phosphor_assets = _root / "sources" / "phosphor-core" / "assets"
else:
phosphor_assets = Path(phosphor_root)
@@ -0,0 +1,122 @@
---
name: migrate_issues_frontmatter
kind: pipeline
lang: py
domain: pipelines
version: "1.0.0"
purity: impure
signature: "def migrate_issues_frontmatter(issues_dir: str, backup_dir: str = '', dry_run: bool = False) -> dict"
description: "Migra dev/issues/*.md de metadata inline (**Key:** value) a frontmatter YAML canonico (issue 0100). Idempotente: archivos ya migrados (id + domain + scope presentes) se saltan. Archivos con frontmatter parcial reciben merge de claves faltantes. Infiere domain, scope, type y priority por heuristicas de nombre de archivo."
tags: [registry-quality, frontmatter, chore, dev-issues, migration, registry]
uses_functions:
- extract_frontmatter_py_core
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports:
- json
- os
- re
- shutil
- sys
- datetime
- pathlib
- typing
tested: false
params:
- name: issues_dir
desc: "Ruta al directorio de issues (ej. 'dev/issues'). Se escanean *.md directos y completed/*.md."
- name: backup_dir
desc: "Directorio donde copiar los originales antes de escribir. Default: <issues_dir>/.backup_pre_0100/. Pasar '' para usar el default."
- name: dry_run
desc: "Si True, calcula cambios pero no escribe archivos ni crea backups."
output: "dict con claves: migrated (N archivos migrados desde cero), merged (N archivos con frontmatter parcial completado), skipped (N ya completos), warnings (lista de strings), files (lista de resultados por archivo con action/domain_inferred/scope_inferred/warnings)."
file_path: "python/functions/pipelines/migrate_issues_frontmatter.py"
notes: |
Pipeline impuro: muta archivos .md en disco y crea backups.
Heuristicas de domain (primer match gana, multi-tag posible):
cpp-stack — filename contiene cpp-, imgui, glfw, glsl, altsnap
kanban — filename contiene kanban
trading — filename contiene trading
gamedev — filename contiene gamedev
osint — filename contiene osint, odr-
data-ingest — filename contiene metabase, bigquery, datafactory, navegator, cdp-
notify — filename contiene notify, telegram, matrix
imagegen — filename contiene imagegen, sd-cpp, stable-diffusion
registry-quality — filename contiene audit-, registry-first, uses_functions, nested-app-md
meta — filename contiene autonomous, e2e-validation, registry-call, delegation, capability, call-monitor, mcp-
deploy — filename contiene deploy, vps
dev-ux — filename contiene fn-run, gradle_run, dev-
browser — filename contiene browser, chrome, cdp-
apps-infra — filename contiene datahub, app-hub, launcher, app-locations
frontend — filename contiene frontend, react
default — lista vacia + warning
Heuristicas de scope:
roadmap -> cross-stack
extract-|migrate-|audit- -> registry-only
type=app o -app- en filename -> app-scoped
default -> multi-app
Heuristicas de type (si no viene del inline):
roadmap -> epic
audit-|cleanup- -> chore
fix-|bugfix- -> bugfix
default -> feature
Heuristicas de priority (si no viene del inline):
mtime < 14 dias -> alta
else -> media
Formatos inline reconocidos:
**Status:** value (y variantes: Estado, Tipo, Prioridad, Created, Depends, Blocks, Related)
| **Campo** | valor | (tabla markdown)
El backup se crea solo UNA vez por archivo (si ya existe el backup, no sobreescribe).
Segunda ejecucion = 0 cambios si los archivos ya tienen id + domain + scope.
---
## Ejemplo
```bash
# Siempre hacer dry-run primero para revisar
cd /home/lucas/fn_registry
python/.venv/bin/python3 python/functions/pipelines/migrate_issues_frontmatter.py dev/issues --dry-run
# Aplicar migracion real
python/.venv/bin/python3 python/functions/pipelines/migrate_issues_frontmatter.py dev/issues
# O via fn run
./fn run migrate_issues_frontmatter_py_pipelines
```
```python
import sys
sys.path.insert(0, "python/functions")
from pipelines.migrate_issues_frontmatter import migrate_issues_frontmatter
# Dry run
result = migrate_issues_frontmatter("dev/issues", dry_run=True)
print(result["migrated"], "to migrate,", result["skipped"], "already done")
# Apply
result = migrate_issues_frontmatter("dev/issues")
print(result["warnings"])
```
## Cuando usarla
Cuando necesites migrar `dev/issues/*.md` de inline `**Key:** value` a frontmatter YAML canonico. Ejecucion one-shot por proyecto; despues de la primera pasada, correr periodicamente para validar issues nuevos (idempotente).
Tambien util al crear un proyecto nuevo que adopte el formato issues/ de fn_registry: copia los .md y lanza el pipeline para homogeneizar.
## Gotchas
- **Muta archivos en disco**. Siempre correr con `--dry-run` primero para revisar la salida.
- **Backup dir no es gitignored por defecto**. Anadir `.backup_pre_0100/` al `.gitignore` del repo si no se quiere versionar.
- **Heuristicas de domain son best-effort**. Revisar warnings al final de la ejecucion; los archivos sin domain inferido quedan con `domain: []` y aparecen en warnings.
- **Frontmatter parcial se completa sin sobreescribir**. Si ya tienes `status: done` en el YAML, no se toca. Solo se anaden las claves ausentes.
- **El backup se escribe antes que el nuevo contenido**. Si el proceso se interrumpe a mitad, los backups son los originales correctos.
- **No valida el schema canonico de los tipos**. Si una issue tiene `type: weird-value`, se guarda tal cual y no hay warning. La validacion es responsabilidad de `fn doctor issues` (issue 0102, pendiente).
@@ -0,0 +1,646 @@
#!/usr/bin/env python3
"""migrate_issues_frontmatter — migrate dev/issues/*.md from inline **Key:** value
metadata to canonical YAML frontmatter (issue 0100).
Idempotent: files that already have id + domain + scope in frontmatter are skipped.
Files with partial frontmatter get missing keys merged in without overwriting existing ones.
"""
from __future__ import annotations
import json
import os
import re
import shutil
import sys
from datetime import date
from pathlib import Path
from typing import Any
# ---------------------------------------------------------------------------
# Registry path setup
# ---------------------------------------------------------------------------
def _find_registry_root() -> Path:
here = Path(__file__).resolve()
for parent in (here, *here.parents):
if (parent / "registry.db").exists():
return parent
return Path.cwd()
_REGISTRY_ROOT = _find_registry_root()
sys.path.insert(0, str(_REGISTRY_ROOT / "python" / "functions"))
from core.core import extract_frontmatter # noqa: E402
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
TODAY = date.today().isoformat()
_SKIP_NAMES = {"README.md", "template.md", "README", "template"}
_STATUS_ALIASES = {
"pendiente": "pendiente",
"pending": "pendiente",
"in-progress": "in-progress",
"en-progreso": "in-progress",
"en_progreso": "in-progress",
"bloqueado": "bloqueado",
"blocked": "bloqueado",
"completado": "completado",
"done": "completado",
"completed": "completado",
"deferred": "deferred",
"diferido": "deferred",
"closed": "completado",
}
_TYPE_ALIASES = {
"app": "app",
"feature": "feature",
"bugfix": "bugfix",
"bug": "bugfix",
"refactor": "refactor",
"chore": "chore",
"docs": "docs",
"doc": "docs",
"spike": "spike",
"epic": "epic",
"infra": "infra",
"planning": "planning",
}
_PRIORITY_ALIASES = {
"alta": "alta",
"high": "alta",
"media": "media",
"medium": "media",
"baja": "baja",
"low": "baja",
}
# ---------------------------------------------------------------------------
# Heuristics
# ---------------------------------------------------------------------------
def _infer_domain(filename: str) -> list[str]:
"""Return list of canonical domain tags based on filename heuristics."""
f = filename.lower()
domains: list[str] = []
if re.search(r"^cpp-|(-cpp-)|imgui|glfw|glsl|altsnap|sizemove", f):
domains.append("cpp-stack")
if re.search(r"^kanban-|kanban", f):
domains.append("kanban")
if re.search(r"^trading-|0088[a-z]?-trading", f):
domains.append("trading")
if re.search(r"^gamedev-|0072[a-z]?-gamedev", f):
domains.append("gamedev")
if re.search(r"osint|odr-", f):
domains.append("osint")
if re.search(r"metabase|bigquery|datafactory|data-factory|navegator|cdp-", f):
domains.append("data-ingest")
if re.search(r"notify|telegram|matrix", f):
domains.append("notify")
if re.search(r"imagegen|sd-cpp|stable-diffusion", f):
domains.append("imagegen")
if re.search(r"dag-engine|dagu", f):
if "cpp" in f or "imgui" in f:
domains.append("cpp-stack")
else:
domains.append("data-ingest")
if re.search(r"audit-|registry-first|uses.functions|nested-app-md", f):
domains.append("registry-quality")
if re.search(r"autonomous|e2e-validation|registry-call|delegation|capability|call-monitor|mcp-", f):
domains.append("meta")
if re.search(r"deploy|vps", f):
domains.append("deploy")
if re.search(r"fn-run|gradle.run|(?<![a-z])dev-(?!_console)", f):
domains.append("dev-ux")
if re.search(r"browser|chrome|cdp-", f):
domains.append("browser")
if re.search(r"datahub|app-hub|launcher|app-locations", f):
domains.append("apps-infra")
if re.search(r"frontend|react", f):
domains.append("frontend")
if re.search(r"0100|frontmatter|migrate-issues|extract-|audit-", f):
if "registry-quality" not in domains:
domains.append("registry-quality")
# deduplicate while preserving order
seen: set[str] = set()
result: list[str] = []
for d in domains:
if d not in seen:
seen.add(d)
result.append(d)
return result
def _infer_scope(filename: str, inline_type: str) -> str:
f = filename.lower()
if re.search(r"roadmap", f):
return "cross-stack"
if re.search(r"extract-|migrate-|audit-", f):
return "registry-only"
if inline_type == "app" or re.search(r"-app[-.]|app-", f):
return "app-scoped"
return "multi-app"
def _infer_type(filename: str) -> str:
f = filename.lower()
if re.search(r"roadmap", f):
return "epic"
if re.search(r"audit-|cleanup-", f):
return "chore"
if re.search(r"fix-|bugfix-|bug-", f):
return "bugfix"
return "feature"
def _infer_priority_from_mtime(path: Path) -> str:
mtime = path.stat().st_mtime
mtime_date = date.fromtimestamp(mtime)
delta = (date.today() - mtime_date).days
if delta <= 14:
return "alta"
return "media"
# ---------------------------------------------------------------------------
# Inline parser
# ---------------------------------------------------------------------------
_H1_RE = re.compile(r"^#\s+(.+)$", re.MULTILINE)
_BOLD_KEY_RE = re.compile(r"^\*\*([A-Za-z]+)\*\*:\s*(.*)$")
_TABLE_META_RE = re.compile(r"^\|\s*\*\*([A-Za-z]+)\*\*\s*\|\s*(.+?)\s*\|")
# Match "NNNN" or "NNNN — rest"
_ID_FROM_H1_RE = re.compile(r"^(\d{4}[a-z]?)\s*[—-]\s*(.+)$")
_ISSUE_REF_RE = re.compile(r"\b(\d{4}[a-z]?)\b")
def _extract_h1_title(content: str) -> tuple[str, str]:
"""Return (issue_id, clean_title) from H1. Falls back to ('', content first line)."""
m = _H1_RE.search(content)
if not m:
return "", ""
h1 = m.group(1).strip()
id_m = _ID_FROM_H1_RE.match(h1)
if id_m:
return id_m.group(1), id_m.group(2).strip()
return "", h1
def _parse_inline_meta(content: str) -> dict[str, str]:
"""Parse **Key:** value lines and table metadata from first ~40 lines."""
meta: dict[str, str] = {}
lines = content.splitlines()[:40]
for line in lines:
# Bold inline: **Status:** value
bm = _BOLD_KEY_RE.match(line.strip())
if bm:
key = bm.group(1).lower()
val = bm.group(2).strip()
meta[key] = val
continue
# Table row: | **Estado** | value |
tm = _TABLE_META_RE.match(line.strip())
if tm:
key = tm.group(1).lower()
val = tm.group(2).strip()
# Strip markdown bold from value
val = re.sub(r"\*\*(.+?)\*\*", r"\1", val)
meta[key] = val
return meta
def _parse_issue_ids(raw: str) -> list[str]:
"""Extract issue IDs (NNNN or NNNNa) from a raw string like '0096, 0097 — DONE'."""
if not raw or raw.strip() in ("", "-", "", "ninguna", "none"):
return []
return _ISSUE_REF_RE.findall(raw)
def _normalize_status(raw: str) -> str:
raw = raw.lower().strip()
# extract first word-token
token = re.split(r"[\s,;—-]", raw)[0]
return _STATUS_ALIASES.get(token, _STATUS_ALIASES.get(raw, "pendiente"))
def _normalize_type(raw: str) -> str:
raw = raw.lower().strip()
# strip trailing annotation like "feature — apps/kanban/"
token = re.split(r"[\s,;—-]", raw)[0]
return _TYPE_ALIASES.get(token, "")
def _normalize_priority(raw: str) -> str:
raw = raw.lower().strip()
token = re.split(r"[\s,;]", raw)[0]
return _PRIORITY_ALIASES.get(token, "")
# ---------------------------------------------------------------------------
# Frontmatter builder
# ---------------------------------------------------------------------------
def _build_frontmatter(
issue_id: str,
title: str,
status: str,
type_: str,
domain: list[str],
scope: str,
priority: str,
depends: list[str],
blocks: list[str],
related: list[str],
created: str,
updated: str,
tags: list[str],
) -> str:
"""Render canonical YAML frontmatter block as a string."""
lines: list[str] = ["---"]
lines.append(f"id: \"{issue_id}\"")
lines.append(f"title: \"{title}\"")
lines.append(f"status: {status}")
lines.append(f"type: {type_}")
if domain:
lines.append("domain:")
for d in domain:
lines.append(f" - {d}")
else:
lines.append("domain: []")
lines.append(f"scope: {scope}")
lines.append(f"priority: {priority}")
if depends:
lines.append("depends:")
for d in depends:
lines.append(f" - \"{d}\"")
else:
lines.append("depends: []")
if blocks:
lines.append("blocks:")
for b in blocks:
lines.append(f" - \"{b}\"")
else:
lines.append("blocks: []")
if related:
lines.append("related:")
for r in related:
lines.append(f" - \"{r}\"")
else:
lines.append("related: []")
lines.append(f"created: {created}")
lines.append(f"updated: {updated}")
if tags:
lines.append("tags:")
for t in tags:
lines.append(f" - {t}")
else:
lines.append("tags: []")
lines.append("---")
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Per-file processing
# ---------------------------------------------------------------------------
FileResult = dict[str, Any]
def _process_file(
path: Path,
backup_dir: Path | None,
dry_run: bool,
) -> FileResult:
result: FileResult = {
"path": str(path),
"action": "skipped",
"domain_inferred": [],
"scope_inferred": "",
"warnings": [],
}
content = path.read_text(encoding="utf-8")
body_without_fm, existing_fm = extract_frontmatter(content)
# Determine issue_id from filename
stem = path.stem # e.g. "0099-datahub-app-launcher"
id_from_file_m = re.match(r"^(\d{4}[a-z]?)", stem)
file_issue_id = id_from_file_m.group(1) if id_from_file_m else stem
# Check if already fully migrated
if existing_fm and isinstance(existing_fm, dict):
fm = existing_fm
has_id = "id" in fm
has_domain = "domain" in fm
has_scope = "scope" in fm
if has_id and has_domain and has_scope:
result["action"] = "skipped"
return result
# Partial frontmatter — merge missing keys
result["action"] = "merged"
_fill_missing_keys(path, fm, body_without_fm, file_issue_id, result, backup_dir, dry_run)
return result
# No frontmatter: parse inline and build canonical
result["action"] = "migrated"
_migrate_from_inline(path, content, file_issue_id, result, backup_dir, dry_run)
return result
def _fill_missing_keys(
path: Path,
fm: dict,
body: str,
file_issue_id: str,
result: FileResult,
backup_dir: Path | None,
dry_run: bool,
) -> None:
"""Merge missing canonical keys into an existing partial frontmatter."""
warnings = result["warnings"]
filename = path.name
issue_id = str(fm.get("id", file_issue_id))
title = str(fm.get("title", _extract_h1_title(body)[1] or filename))
status = _normalize_status(str(fm.get("status", fm.get("estado", "pendiente"))))
type_ = _normalize_type(str(fm.get("type", fm.get("tipo", ""))))
if not type_:
type_ = _infer_type(filename)
if not type_:
warnings.append(f"type missing, inferred: {type_ or '(empty)'}")
domain_raw = fm.get("domain", fm.get("dominio", []))
if isinstance(domain_raw, str):
domain = [domain_raw] if domain_raw else []
elif isinstance(domain_raw, list):
domain = domain_raw
else:
domain = []
if not domain:
domain = _infer_domain(filename)
result["domain_inferred"] = domain
if not domain:
warnings.append("domain could not be inferred")
scope = str(fm.get("scope", ""))
if not scope:
scope = _infer_scope(filename, type_)
result["scope_inferred"] = scope
priority = _normalize_priority(str(fm.get("priority", fm.get("priority", ""))))
if not priority:
priority = _infer_priority_from_mtime(path)
warnings.append(f"priority missing, inferred from mtime: {priority}")
depends = _coerce_id_list(fm.get("depends", fm.get("depends_on", [])))
blocks = _coerce_id_list(fm.get("blocks", []))
related = _coerce_id_list(fm.get("related", []))
created = str(fm.get("created", TODAY))
tags_raw = fm.get("tags", [])
tags = tags_raw if isinstance(tags_raw, list) else []
new_fm = _build_frontmatter(
issue_id=issue_id,
title=title,
status=status,
type_=type_,
domain=domain,
scope=scope,
priority=priority,
depends=depends,
blocks=blocks,
related=related,
created=created,
updated=TODAY,
tags=tags,
)
new_content = new_fm + body
if not dry_run:
_backup_file(path, backup_dir)
path.write_text(new_content, encoding="utf-8")
def _coerce_id_list(val: Any) -> list[str]:
if isinstance(val, list):
return [str(v) for v in val]
if isinstance(val, str):
return _parse_issue_ids(val)
return []
def _migrate_from_inline(
path: Path,
content: str,
file_issue_id: str,
result: FileResult,
backup_dir: Path | None,
dry_run: bool,
) -> None:
"""Parse inline metadata and write YAML frontmatter."""
warnings = result["warnings"]
filename = path.name
inline = _parse_inline_meta(content)
h1_id, h1_title = _extract_h1_title(content)
issue_id = h1_id or file_issue_id
title = h1_title or filename
# Status
status_raw = inline.get("status", inline.get("estado", ""))
status = _normalize_status(status_raw) if status_raw else "pendiente"
# Type
type_raw = inline.get("type", inline.get("tipo", ""))
type_ = _normalize_type(type_raw) if type_raw else ""
if not type_:
type_ = _infer_type(filename)
warnings.append(f"type missing/unknown ({type_raw!r}), inferred: {type_}")
# Domain
domain_raw = inline.get("domain", inline.get("dominio", ""))
if domain_raw and domain_raw not in ("", "-"):
domain = [d.strip() for d in re.split(r"[,;]", domain_raw) if d.strip()]
else:
domain = _infer_domain(filename)
result["domain_inferred"] = domain
if not domain:
warnings.append("domain could not be inferred from filename")
# Scope
scope_raw = inline.get("scope", inline.get("alcance", ""))
scope = scope_raw if scope_raw and scope_raw not in ("", "-") else _infer_scope(filename, type_)
result["scope_inferred"] = scope
# Priority
priority_raw = inline.get("priority", inline.get("prioridad", ""))
priority = _normalize_priority(priority_raw) if priority_raw else ""
if not priority:
priority = _infer_priority_from_mtime(path)
warnings.append(f"priority missing, inferred from mtime: {priority}")
# Depends / Blocks / Related
depends = _parse_issue_ids(inline.get("depends", inline.get("depends_on", inline.get("depende", ""))))
blocks = _parse_issue_ids(inline.get("blocks", inline.get("bloquea", "")))
related = _parse_issue_ids(inline.get("related", inline.get("relacionado", inline.get("relacionados", ""))))
# Created date
created_raw = inline.get("created", inline.get("fecha", ""))
created = created_raw.strip() if created_raw and re.match(r"\d{4}-\d{2}-\d{2}", created_raw) else TODAY
new_fm = _build_frontmatter(
issue_id=issue_id,
title=title,
status=status,
type_=type_,
domain=domain,
scope=scope,
priority=priority,
depends=depends,
blocks=blocks,
related=related,
created=created,
updated=TODAY,
tags=[],
)
new_content = new_fm + content
if not dry_run:
_backup_file(path, backup_dir)
path.write_text(new_content, encoding="utf-8")
def _backup_file(path: Path, backup_dir: Path | None) -> None:
if backup_dir is None:
return
backup_dir.mkdir(parents=True, exist_ok=True)
dest = backup_dir / path.name
if not dest.exists():
shutil.copy2(path, dest)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def migrate_issues_frontmatter(
issues_dir: str,
backup_dir: str = "",
dry_run: bool = False,
) -> dict[str, Any]:
"""Migrate dev/issues/*.md from inline metadata to canonical YAML frontmatter.
Args:
issues_dir: Path to the issues directory (e.g. "dev/issues").
backup_dir: Where to copy originals before writing. Defaults to
<issues_dir>/.backup_pre_0100/. Pass "" for default.
dry_run: If True, compute changes but do not write files.
Returns:
{
"migrated": N,
"merged": N,
"skipped": N,
"warnings": [...],
"files": [{"path": ..., "action": ..., "domain_inferred": ..., "scope_inferred": ..., "warnings": [...]}, ...]
}
"""
issues_path = Path(issues_dir).expanduser()
if not issues_path.is_dir():
raise FileNotFoundError(f"issues_dir not found: {issues_path}")
# Determine backup directory
if dry_run:
bk_path = None
elif backup_dir:
bk_path = Path(backup_dir).expanduser()
else:
bk_path = issues_path / ".backup_pre_0100"
# Collect all .md files from issues_dir and issues_dir/completed/
all_files: list[Path] = []
for md in sorted(issues_path.glob("*.md")):
if md.name in _SKIP_NAMES:
continue
all_files.append(md)
completed_dir = issues_path / "completed"
if completed_dir.is_dir():
for md in sorted(completed_dir.glob("*.md")):
if md.name in _SKIP_NAMES:
continue
all_files.append(md)
results: list[FileResult] = []
all_warnings: list[str] = []
for path in all_files:
try:
r = _process_file(path, bk_path, dry_run)
except Exception as e:
r = {
"path": str(path),
"action": "warning",
"domain_inferred": [],
"scope_inferred": "",
"warnings": [f"ERROR: {e}"],
}
if r["warnings"]:
for w in r["warnings"]:
all_warnings.append(f"{path.name}: {w}")
results.append(r)
migrated = sum(1 for r in results if r["action"] == "migrated")
merged = sum(1 for r in results if r["action"] == "merged")
skipped = sum(1 for r in results if r["action"] == "skipped")
warnings_count = sum(1 for r in results if r["action"] == "warning")
# Summary table
prefix = "[DRY RUN] " if dry_run else ""
print(f"\n{prefix}migrate_issues_frontmatter summary")
print(f" issues_dir : {issues_path}")
print(f" files found: {len(results)}")
print(f" migrated : {migrated}")
print(f" merged : {merged}")
print(f" skipped : {skipped}")
print(f" warnings : {len(all_warnings)}")
if all_warnings:
print("\nWarnings:")
for w in all_warnings:
print(f" - {w}")
return {
"migrated": migrated,
"merged": merged,
"skipped": skipped,
"warnings": all_warnings,
"files": results,
}
# ---------------------------------------------------------------------------
# CLI entry
# ---------------------------------------------------------------------------
if __name__ == "__main__":
issues_dir_arg = sys.argv[1] if len(sys.argv) > 1 else "dev/issues"
dry = "--dry-run" in sys.argv
result = migrate_issues_frontmatter(issues_dir_arg, dry_run=dry)
print(json.dumps(result, ensure_ascii=False, indent=2))